How to isolate context, chat history, and workflow checkpoints per user when using Workflow / WorkflowAgent with ChatKitServer?

[VanQuy24]

I am integrating Agent Framework Workflows with ChatKitServer where multiple users interact with the same WorkflowAgent instance.
My workflow uses multiple executors and also relies on HITL (ctx.request_info) at certain steps.

I need guidance on the correct approach to managing context, chat history, and workflow checkpointing per user.

---

Architecture

• ChatKitServer calls the same WorkflowAgent instance for all users.
• The Workflow is composed of several executors.
• I use ctx.shared_state to carry data through the workflow.
• Some steps require HITL via ctx.request_info.

---

Questions

1. How to isolate context between different users?

Right now, conversations from different users appear to overwrite or leak into each other, especially when workflows run concurrently.
I need a recommended approach to ensure each user gets clean and independent workflow state.

2. How to store chat history and workflow checkpoints properly?

My goal is that when a user returns later (possibly hours or days later), the system should be able to:

• Resume the correct workflow state
• Restore the correct conversation history
• Continue any pending HITL state (e.g., waiting for user input)
• Avoid mixing with other users’ conversations

I want to understand the best practices for combining WorkflowAgent + ChatKitServer so that each user’s session is recoverable and isolated.

3. Where should I store context and state?

I am unsure what is the recommended place to persist:

• Conversational history
• Shared state (ctx.shared_state)
• Workflow internal state / checkpoints
• HITL waiting state

Should these be stored:

• In ChatKit’s conversation/message store?
• In an external DB?
• Using Agent Framework’s built-in state provider?
• Or is there a supported pattern for plugging in a custom state backend?

Any guidance or patterns from the maintainers is appreciated.

---

Issue I’m facing

When multiple users interact with the system at the same time:

• Their context overwrites each other
• Shared state is not scoped to a user
• HITL steps become confused or mismatched
• The WorkflowAgent behaves as if it has a global session, not one per user

I believe I’m missing a recommended approach for user-scoped state management when combining WorkflowAgent + ChatKitServer.

---

What I hope to learn

• How to correctly structure user-scoped workflows
• How to persist chat history and workflow state per user
• How to handle HITL steps safely in multi-user environments
• Best practices for integrating WorkflowAgent with ChatKitServer

Any official guidance, sample patterns, or architectural recommendations would be extremely helpful.

[moonbox3]

Hi @VanQuy24,

How to correctly structure user-scoped workflows

Use thread.id as the session key. Create per-user CheckpointStorage instances.

How to persist chat history and workflow state per user

Persisting chat history and workflow state:

• Chat history: ChatKit's Store handles this automatically
• Workflow state: Use FileCheckpointStorage per user

How to handle HITL steps safely in multi-user environments

Checkpoint captures pending_request_info_events. Resume via checkpoint_id parameter - the user's next input fulfills the pending request.

Best practices for integrating WorkflowAgent with ChatKitServer

An example of this could be for HITL:

from agent_framework import FileCheckpointStorage
from agent_framework.chatkit import stream_agent_response

class MultiUserChatKitServer(ChatKitServer[dict[str, Any]]):
    def __init__(self, store, workflow):
        super().__init__(store=store)
        self.workflow = workflow
        self._user_checkpoints: dict[str, FileCheckpointStorage] = {}

    async def respond(self, thread: ThreadMetadata, input_user_message: UserMessageItem | None, context: dict[str, Any]):
        user_id = thread.id
        
        # Per-user checkpoint storage (use FileCheckpointStorage for persistence)
        if user_id not in self._user_checkpoints:
            self._user_checkpoints[user_id] = FileCheckpointStorage(f"./checkpoints/{user_id}")
        
        checkpoint_storage = self._user_checkpoints[user_id]
        agent = self.workflow.as_agent()
        
        # Resume from latest checkpoint if exists (handles HITL state too)
        checkpoints = await checkpoint_storage.list_checkpoints()
        checkpoint_id = checkpoints[-1].checkpoint_id if checkpoints else None
        
        async for event in stream_agent_response(
            agent.run_stream(messages, checkpoint_id=checkpoint_id, checkpoint_storage=checkpoint_storage),
            thread_id=thread.id,
        ):
            yield event