Python Sharing
This guide shows how to add conversation sharing and ownership transfer APIs to the Python tutorial app.
New to sharing concepts? Read Sharing & Access Control first to understand access levels and transfer flow.
Prerequisites
Starting checkpoint: This guide starts from python/examples/langchain/doc-checkpoints/03-with-history
Make sure you’ve completed:
- Python Conversation History - History recording and conversation APIs
- Multiple users in Keycloak (
bob,alice,charlie)
Also complete Step 2 in Python Dev Setup (build local memory-service-langchain wheel + UV_FIND_LINKS); this is temporary until the package is released.
Membership Endpoints
Checkpoint 06 adds membership endpoint passthroughs to the tutorial app:
response = await proxy.list_memberships(conversation_id)
return to_fastapi_response(response)
@app.post("/v1/conversations/{conversation_id}/memberships")
async def create_membership(conversation_id: str, request: Request):
response = await proxy.create_membership(conversation_id, await parse_optional_json(request) or {})
return to_fastapi_response(response)
@app.patch("/v1/conversations/{conversation_id}/memberships/{user_id}")
async def update_membership(conversation_id: str, user_id: str, request: Request):
response = await proxy.update_membership(
conversation_id,
user_id,
await parse_optional_json(request) or {},
)
return to_fastapi_response(response)
@app.delete("/v1/conversations/{conversation_id}/memberships/{user_id}")
async def delete_membership(conversation_id: str, user_id: str):
response = await proxy.delete_membership(conversation_id, user_id)
return to_fastapi_response(response)
Why: These endpoints are thin pass-throughs so the agent app does not need to duplicate any membership logic. The MemoryServiceProxy injects the agent API key header automatically, while the user’s Bearer token (forwarded by install_fastapi_authorization_middleware) provides identity for access-level enforcement on the Memory Service side. parse_optional_json is used on write operations to gracefully handle requests with empty bodies.
Ownership Transfer Endpoints
Checkpoint 06 also exposes ownership transfer APIs:
response = await proxy.list_ownership_transfers(
role=request.query_params.get("role"),
after_cursor=request.query_params.get("afterCursor"),
limit=int(limit) if (limit := request.query_params.get("limit")) is not None else None,
)
return to_fastapi_response(response)
@app.post("/v1/ownership-transfers")
async def create_ownership_transfer(request: Request):
response = await proxy.create_ownership_transfer(await parse_optional_json(request) or {})
return to_fastapi_response(response)
@app.post("/v1/ownership-transfers/{transfer_id}/accept")
async def accept_ownership_transfer(transfer_id: str):
response = await proxy.accept_ownership_transfer(transfer_id)
return to_fastapi_response(response)
@app.delete("/v1/ownership-transfers/{transfer_id}")
async def delete_ownership_transfer(transfer_id: str):
response = await proxy.delete_ownership_transfer(transfer_id)
return to_fastapi_response(response) Why: Ownership transfer is a two-step handshake: the current owner creates a pending transfer, and the recipient accepts it, which atomically promotes the recipient to owner and demotes the previous owner to manager. Exposing list and delete allows both parties to view outstanding transfers and to cancel or decline them. The DELETE endpoint uses parse_optional_json because HTTP DELETE requests from some clients may omit a body entirely.
Testing Membership Management
Define a token helper for multiple users:
function get-token() {
local username=${1:-bob}
local password=${2:-$username}
curl -sSfX POST http://localhost:8081/realms/memory-service/protocol/openid-connect/token \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "client_id=memory-service-client" \
-d "client_secret=change-me" \
-d "grant_type=password" \
-d "username=$username" \
-d "password=$password" \
| jq -r '.access_token'
}
Create a conversation as bob:
curl -sSfX POST http://localhost:9090/chat/a37bf20b-34eb-4471-b481-107c68ac3a9d \
-H "Authorization: Bearer $(get-token bob bob)" \
-H "Content-Type: text/plain" \
-d "Hello, starting a sharing test conversation." Example output:
I am a Python memory-service demo agent. List memberships:
curl -sSfX GET http://localhost:9090/v1/conversations/a37bf20b-34eb-4471-b481-107c68ac3a9d/memberships \
-H "Authorization: Bearer $(get-token bob bob)" | jq Example output:
{
"afterCursor": null,
"data": [
{
"conversationId": "a37bf20b-34eb-4471-b481-107c68ac3a9d",
"userId": "bob",
"accessLevel": "owner",
"createdAt": "2026-03-06T09:05:28.492968-05:00"
}
]
} Add alice as a writer:
curl -sSfX POST http://localhost:9090/v1/conversations/a37bf20b-34eb-4471-b481-107c68ac3a9d/memberships \
-H "Authorization: Bearer $(get-token bob bob)" \
-H "Content-Type: application/json" \
-d '{"userId":"alice","accessLevel":"writer"}' | jq Example output:
{
"conversationId": "a37bf20b-34eb-4471-b481-107c68ac3a9d",
"userId": "alice",
"accessLevel": "writer",
"createdAt": "2026-03-06T09:05:28.670625-05:00"
} Testing Ownership Transfers
Initiate an ownership transfer:
curl -sSfX POST http://localhost:9090/v1/ownership-transfers \
-H "Authorization: Bearer $(get-token bob bob)" \
-H "Content-Type: application/json" \
-d '{
"conversationId": "a37bf20b-34eb-4471-b481-107c68ac3a9d",
"newOwnerUserId": "alice"
}' | jq
List transfers for the recipient:
curl -sSfX GET "http://localhost:9090/v1/ownership-transfers?role=recipient" \
-H "Authorization: Bearer $(get-token alice alice)" | jq
Accept a transfer:
curl -sSfX POST http://localhost:9090/v1/ownership-transfers/<transfer-id>/accept \
-H "Authorization: Bearer $(get-token alice alice)"
Decline or cancel a transfer:
curl -sSfX DELETE http://localhost:9090/v1/ownership-transfers/<transfer-id> \
-H "Authorization: Bearer $(get-token alice alice)"
Completed Checkpoint
Completed code: View the full implementation at python/examples/langchain/doc-checkpoints/06-sharing
What’s Next
You’ve now added collaborative features (memberships and ownership transfer) to your Python tutorial app.
Next, continue to Indexing and Search to add searchable indexed content.