agent/agent-framework
1
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: feat: Add ChatKit integration with a sample application (#1273)

Browse Source 
* feat: Add ChatKit integration with a new frontend application

- Created a new frontend application using React and Vite for the ChatKit integration.
- Added essential files including package.json, vite.config.ts, and Tailwind CSS configuration.
- Implemented core components: App, Home, ChatKitPanel, ThemeToggle, and hooks for color scheme management.
- Established SQLite-based store implementation for ChatKit data persistence in store.py.
- Integrated theme toggling functionality for light and dark modes.
- Set up ESLint and TypeScript configurations for better development experience.

* git ignore

* fix mypy

* add mising file

* minimal frontend for chatkit sample

* update ignore files

* version

* set python version lowerbound on chatkit

* update project settings for chatkit

* update setup

* update setup

* update setup

* update setup

* weather widget

* add select city widget sample

* remove widget helper

* update chatkit to include file attachments and cover more thread item types

* update readme with mermaid diagram

* update diagram

* update instructions

* update chatkit dependency

* fix converter imports

* move to demos/

* move to demos/ -- rename references

* support multiple session instead of using global variable in sample

* support chunk streaming

* fix tests

* Update python/samples/demos/chatkit-integration/store.py

Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>

* use local host

---------

Co-authored-by: Evan Mattson <35585003+moonbox3@users.noreply.github.com>
This commit is contained in:
Eric Zhu
2025-11-04 18:11:40 -08:00
committed by GitHub
Unverified
parent bbde248839
commit 0c862e97a6
33 changed files with 4972 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+9 -1
View File
@@ -203,4 +203,12 @@ agents.md
# AI
.claude/
WARP.md
WARP.md
# Frontend
**/frontend/node_modules/
**/frontend/.vite/
**/frontend/dist/
# Database files
*.db
python/.pre-commit-config.yaml
+1 -1
View File
@@ -14,7 +14,7 @@ repos:
- id: check-json
name: Check JSON files
files: \.json$
exclude: ^.*\.vscode\/.*
exclude: ^.*\.vscode\/.*|^python/demos/samples/chatkit-integration/frontend/(tsconfig.*\.json|package-lock\.json)$
- id: end-of-file-fixer
name: Fix End of File
files: \.py$
python/packages/chatkit/.gitignore
+3
View File
@@ -0,0 +1,3 @@
chatkit-python
openai-chatkit-advanced-samples
chatkit-js
python/packages/chatkit/LICENSE
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) Microsoft Corporation.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
python/packages/chatkit/README.md
+87
View File
@@ -0,0 +1,87 @@
# Agent Framework and ChatKit Integration
This package provides an integration layer between Microsoft Agent Framework
and [OpenAI ChatKit (Python)](https://github.com/openai/chatkit-python/).
Specifically, it mirrors the [Agent SDK integration](https://github.com/openai/chatkit-python/blob/main/docs/server.md#agents-sdk-integration), and provides the following helpers:
- `stream_agent_response`: A helper to convert a streamed `AgentRunResponseUpdate`
from a Microsoft Agent Framework agent that implements `AgentProtocol` to ChatKit events.
- `ThreadItemConverter`: A extendable helper class to convert ChatKit thread items to
`ChatMessage` objects that can be consumed by an Agent Framework agent.
- `simple_to_agent_input`: A helper function that uses the default implementation
of `ThreadItemConverter` to convert a ChatKit thread to a list of `ChatMessage`,
useful for getting started quickly.
## Installation
```bash
pip install agent-framework-chatkit --pre
```
This will install `agent-framework-core` and `openai-chatkit` as dependencies.
## Example Usage
Here's a minimal example showing how to integrate Agent Framework with ChatKit:
```python
from collections.abc import AsyncIterator
from typing import Any
from azure.identity import AzureCliCredential
from fastapi import FastAPI, Request
from fastapi.responses import Response, StreamingResponse
from agent_framework import ChatAgent
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.chatkit import simple_to_agent_input, stream_agent_response
from chatkit.server import ChatKitServer
from chatkit.types import ThreadMetadata, UserMessageItem, ThreadStreamEvent
# You'll need to implement a Store - see the sample for a SQLiteStore implementation
from your_store import YourStore # type: ignore[import-not-found] # Replace with your Store implementation
# Define your agent with tools
agent = ChatAgent(
chat_client=AzureOpenAIChatClient(credential=AzureCliCredential()),
instructions="You are a helpful assistant.",
tools=[], # Add your tools here
)
# Create a ChatKit server that uses your agent
class MyChatKitServer(ChatKitServer[dict[str, Any]]):
async def respond(
self,
thread: ThreadMetadata,
input_user_message: UserMessageItem | None,
context: dict[str, Any],
) -> AsyncIterator[ThreadStreamEvent]:
if input_user_message is None:
return
# Convert ChatKit message to Agent Framework format
agent_messages = await simple_to_agent_input(input_user_message)
# Run the agent and stream responses
response_stream = agent.run_stream(agent_messages)
# Convert agent responses back to ChatKit events
async for event in stream_agent_response(response_stream, thread.id):
yield event
# Set up FastAPI endpoint
app = FastAPI()
chatkit_server = MyChatKitServer(YourStore()) # type: ignore[misc]
@app.post("/chatkit")
async def chatkit_endpoint(request: Request):
result = await chatkit_server.process(await request.body(), {"request": request})
if hasattr(result, '__aiter__'): # Streaming
return StreamingResponse(result, media_type="text/event-stream") # type: ignore[arg-type]
else: # Non-streaming
return Response(content=result.json, media_type="application/json") # type: ignore[union-attr]
```
For a complete end-to-end example with a full frontend, see the [weather agent sample](../../samples/demos/chatkit-integration/README.md).
python/packages/chatkit/agent_framework_chatkit/__init__.py
+25
View File
@@ -0,0 +1,25 @@
# Copyright (c) Microsoft. All rights reserved.
"""Agent Framework and ChatKit Integration.
This package provides an integration layer between Microsoft Agent Framework
and OpenAI ChatKit (Python). It mirrors the Agent SDK integration and provides
helpers to convert between Agent Framework and ChatKit types.
"""
import importlib.metadata
from ._converter import ThreadItemConverter, simple_to_agent_input
from ._streaming import stream_agent_response
try:
__version__ = importlib.metadata.version(__name__)
except importlib.metadata.PackageNotFoundError:
__version__ = "0.0.0" # Fallback for development mode
__all__ = [
"ThreadItemConverter",
"__version__",
"simple_to_agent_input",
"stream_agent_response",
]
python/packages/chatkit/agent_framework_chatkit/_converter.py
+603
View File
@@ -0,0 +1,603 @@
# Copyright (c) Microsoft. All rights reserved.
"""Converter utilities for converting ChatKit thread items to Agent Framework messages."""
import logging
import sys
from collections.abc import Awaitable, Callable, Sequence
if sys.version_info >= (3, 11):
from typing import assert_never
else:
from typing_extensions import assert_never
from agent_framework import (
ChatMessage,
DataContent,
FunctionCallContent,
FunctionResultContent,
Role,
TextContent,
UriContent,
)
from chatkit.types import (
AssistantMessageItem,
Attachment,
ClientToolCallItem,
EndOfTurnItem,
HiddenContextItem,
ImageAttachment,
TaskItem,
ThreadItem,
UserMessageItem,
UserMessageTagContent,
UserMessageTextContent,
WidgetItem,
WorkflowItem,
)
logger = logging.getLogger(__name__)
class ThreadItemConverter:
"""Helper class to convert ChatKit thread items to Agent Framework ChatMessage objects.
This class provides a base implementation for converting ChatKit thread items
to Agent Framework messages. It can be extended to handle attachments,
@-mentions, hidden context items, and custom thread item formats.
Args:
attachment_data_fetcher: Optional async function to fetch attachment binary data.
If provided, it should take an attachment ID and return the binary data as bytes.
If not provided, attachments will be converted to UriContent using available URLs.
"""
def __init__(
self,
attachment_data_fetcher: Callable[[str], Awaitable[bytes]] | None = None,
) -> None:
"""Initialize the converter.
Args:
attachment_data_fetcher: Optional async function to fetch attachment data by ID.
"""
self.attachment_data_fetcher = attachment_data_fetcher
async def user_message_to_input(
self, item: UserMessageItem, is_last_message: bool = True
) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit UserMessageItem to Agent Framework ChatMessage(s).
This method is called internally by `to_agent_input()`. Override this method
to customize how user messages are converted.
Args:
item: The ChatKit user message item to convert.
is_last_message: Whether this is the last message in the thread (used for quoted_text handling).
Returns:
A ChatMessage, list of messages, or None to skip.
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types and provides proper message ordering.
"""
# Extract text content from the user message
text_content = ""
if item.content:
for content_part in item.content:
if isinstance(content_part, UserMessageTextContent):
text_content += content_part.text
# Convert attachments to DataContent or UriContent
data_contents: list[DataContent | UriContent] = []
if item.attachments:
for attachment in item.attachments:
content = await self.attachment_to_message_content(attachment)
if content is not None:
data_contents.append(content)
# Create the message with text and attachments
if not text_content.strip() and not data_contents:
return None
# If only text and no attachments, use text parameter for simplicity
if text_content.strip() and not data_contents:
user_message = ChatMessage(role=Role.USER, text=text_content.strip())
else:
# Build contents list with both text and attachments
contents: list[TextContent | DataContent | UriContent] = []
if text_content.strip():
contents.append(TextContent(text=text_content.strip()))
contents.extend(data_contents)
user_message = ChatMessage(role=Role.USER, contents=contents)
# Handle quoted text if this is the last message
messages = [user_message]
if item.quoted_text and is_last_message:
quoted_context = ChatMessage(
role=Role.USER,
text=f"The user is referring to this in particular:\n{item.quoted_text}",
)
# Prepend quoted context before the main message
messages.insert(0, quoted_context)
return messages
async def attachment_to_message_content(self, attachment: Attachment) -> DataContent | UriContent | None:
"""Convert a ChatKit attachment to Agent Framework content.
This method is called internally by `user_message_to_input()` to handle attachments.
Override this method to customize attachment handling for your storage backend.
The default implementation provides two strategies:
1. If an attachment_data_fetcher was provided, it fetches the binary data
and creates a DataContent object
2. Otherwise, for ImageAttachment with preview_url, it creates a UriContent object
For FileAttachment without a data fetcher, returns None (attachment is skipped).
Args:
attachment: The ChatKit attachment to convert (FileAttachment or ImageAttachment).
Returns:
DataContent if binary data is available, UriContent if only URL is available,
or None if the attachment cannot be converted.
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types including attachments within user messages.
Examples:
.. code-block:: python
# With data fetcher
async def fetch_data(attachment_id: str) -> bytes:
return await my_storage.get_file(attachment_id)
converter = ThreadItemConverter(attachment_data_fetcher=fetch_data)
messages = await converter.to_agent_input(thread_items)
# Without data fetcher (uses URLs for images)
converter = ThreadItemConverter()
messages = await converter.to_agent_input(thread_items)
"""
# If we have a data fetcher, use it to get binary data
if self.attachment_data_fetcher is not None:
try:
data = await self.attachment_data_fetcher(attachment.id)
return DataContent(data=data, media_type=attachment.mime_type)
except Exception as e:
# If fetch fails, fall through to URL-based approach
logger.debug(f"Failed to fetch attachment data for {attachment.id}: {e}")
# For ImageAttachment, try to use preview_url
if isinstance(attachment, ImageAttachment) and attachment.preview_url:
return UriContent(uri=str(attachment.preview_url), media_type=attachment.mime_type)
# For FileAttachment without data fetcher, skip the attachment
# Subclasses can override this method to provide custom handling
return None
def hidden_context_to_input(self, item: HiddenContextItem) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit HiddenContextItem to Agent Framework ChatMessage(s).
This method is called internally by `to_agent_input()`. Override this method
to customize how hidden context is converted.
The default implementation wraps the hidden context in XML tags and returns
a system message. This allows the model to distinguish hidden context from
regular conversation.
Args:
item: The ChatKit hidden context item to convert.
Returns:
A ChatMessage with system role, a list of messages, or None to skip.
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types and provides proper message ordering.
Examples:
.. code-block:: python
# Default behavior
converter = ThreadItemConverter()
hidden_item = HiddenContextItem(
id="ctx_1",
thread_id="thread_1",
created_at=datetime.now(),
content="User's email: user@example.com",
)
message = converter.hidden_context_to_input(hidden_item)
# Returns: ChatMessage(role=SYSTEM, text="<HIDDEN_CONTEXT>User's email: ...</HIDDEN_CONTEXT>")
"""
return ChatMessage(role=Role.SYSTEM, text=f"<HIDDEN_CONTEXT>{item.content}</HIDDEN_CONTEXT>")
def tag_to_message_content(self, tag: UserMessageTagContent) -> TextContent:
"""Convert a ChatKit tag (@-mention) to Agent Framework content.
This method is called internally by `user_message_to_input()` to handle tags.
Override this method to customize tag conversion for your application.
The default implementation extracts the tag's display name and wraps it in
XML tags to provide context to the model about the @-mention.
Args:
tag: The ChatKit tag content to convert.
Returns:
TextContent with the tag information.
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types including tags within user messages.
Examples:
.. code-block:: python
# Default behavior
converter = ThreadItemConverter()
tag = UserMessageTagContent(
type="input_tag", id="tag_1", text="john", data={"name": "John Doe"}, interactive=False
)
content = converter.tag_to_message_content(tag)
# Returns: TextContent(text="<TAG>Name:John Doe</TAG>")
"""
name = getattr(tag.data, "name", tag.text if hasattr(tag, "text") else "unknown")
return TextContent(text=f"<TAG>Name:{name}</TAG>")
def task_to_input(self, item: TaskItem) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit TaskItem to Agent Framework ChatMessage(s).
This method is called internally by `to_agent_input()`. Override this method
to customize how tasks are converted.
The default implementation converts custom tasks with title/content into
a user message explaining what task was displayed to the user.
Args:
item: The ChatKit task item to convert.
Returns:
A ChatMessage, a list of messages, or None to skip the task.
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types and provides proper message ordering.
Examples:
.. code-block:: python
# Task with both title and content
from chatkit.types import Task
task_item = TaskItem(
id="task_1",
thread_id="thread_1",
created_at=datetime.now(),
task=Task(type="custom", title="Data Analysis", content="Analyzed sales data"),
)
message = converter.task_to_input(task_item)
# Returns message explaining the task was performed
"""
if item.task.type != "custom" or (not item.task.title and not item.task.content):
return None
title = item.task.title or ""
content = item.task.content or ""
task_text = f"{title}: {content}" if title and content else title or content
text = (
f"A message was displayed to the user that the following task was performed:\n<Task>\n{task_text}\n</Task>"
)
return ChatMessage(role=Role.USER, text=text)
def workflow_to_input(self, item: WorkflowItem) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit WorkflowItem to Agent Framework ChatMessage(s).
This method is called internally by `to_agent_input()`. Override this method
to customize how workflows are converted.
The default implementation converts each custom task in the workflow into
a separate user message explaining what tasks were performed.
Args:
item: The ChatKit workflow item to convert.
Returns:
A list of ChatMessages (one per task), a single message, or None to skip.
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types and provides proper message ordering.
Examples:
.. code-block:: python
# Workflow with multiple tasks
from chatkit.types import Workflow, Task
workflow_item = WorkflowItem(
id="wf_1",
thread_id="thread_1",
created_at=datetime.now(),
workflow=Workflow(
type="custom",
tasks=[
Task(type="custom", title="Step 1", content="Gathered data"),
Task(type="custom", title="Step 2", content="Analyzed results"),
],
),
)
messages = converter.workflow_to_input(workflow_item)
# Returns list of messages for each task
"""
messages: list[ChatMessage] = []
for task in item.workflow.tasks:
if task.type != "custom" or (not task.title and not task.content):
continue
title = task.title or ""
content = task.content or ""
task_text = f"{title}: {content}" if title and content else title or content
text = (
"A message was displayed to the user that the following task was performed:\n"
f"<Task>\n{task_text}\n</Task>"
)
messages.append(ChatMessage(role=Role.USER, text=text))
return messages if messages else None
def widget_to_input(self, item: WidgetItem) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit WidgetItem to Agent Framework ChatMessage(s).
This method is called internally by `to_agent_input()`. Override this method
to customize how widgets are converted.
The default implementation converts the widget to a JSON representation
and includes it in a user message, allowing the model to understand what
UI element was displayed to the user.
Args:
item: The ChatKit widget item to convert.
Returns:
A ChatMessage describing the widget, or None to skip.
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types and provides proper message ordering.
Examples:
.. code-block:: python
# Widget item
from chatkit.widgets import Card, Text
widget_item = WidgetItem(
id="widget_1",
thread_id="thread_1",
created_at=datetime.now(),
widget=Card(children=[Text(value="Hello")]),
)
message = converter.widget_to_input(widget_item)
# Returns message with JSON representation of the widget
"""
try:
widget_json = item.widget.model_dump_json(exclude_unset=True, exclude_none=True)
text = f"The following graphical UI widget (id: {item.id}) was displayed to the user:{widget_json}"
return ChatMessage(role=Role.USER, text=text)
except Exception:
# If JSON serialization fails, skip the widget
return None
async def assistant_message_to_input(self, item: AssistantMessageItem) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit AssistantMessageItem to Agent Framework ChatMessage(s).
The default implementation extracts text from all content parts and creates
an assistant message.
Args:
item: The ChatKit assistant message item to convert.
Returns:
A ChatMessage with assistant role, or None to skip.
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types and provides proper message ordering.
"""
# Extract text from all content parts
text_parts = [content.text for content in item.content]
if not text_parts:
return None
return ChatMessage(role=Role.ASSISTANT, text="".join(text_parts))
async def client_tool_call_to_input(self, item: ClientToolCallItem) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit ClientToolCallItem to Agent Framework ChatMessage(s).
The default implementation converts completed tool calls into function call
and result content.
Args:
item: The ChatKit client tool call item to convert.
Returns:
A list containing function call and result messages, or None for pending calls.
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types and provides proper message ordering.
"""
if item.status == "pending":
# Skip pending tool calls - they cannot be sent to the model
return None
import json
# Create function call message
function_call_msg = ChatMessage(
role=Role.ASSISTANT,
contents=[
FunctionCallContent(
call_id=item.call_id,
name=item.name,
arguments=json.dumps(item.arguments),
)
],
)
# Create function result message
function_result_msg = ChatMessage(
role=Role.TOOL,
contents=[
FunctionResultContent(
call_id=item.call_id,
result=json.dumps(item.output) if item.output is not None else "",
)
],
)
return [function_call_msg, function_result_msg]
async def end_of_turn_to_input(self, item: EndOfTurnItem) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit EndOfTurnItem to Agent Framework ChatMessage(s).
The default implementation skips end-of-turn markers as they are only UI hints.
Args:
item: The ChatKit end-of-turn item to convert.
Returns:
None (end-of-turn items are not converted).
Note:
Instead of calling this method directly, use `to_agent_input()` which handles
all ThreadItem types and provides proper message ordering.
"""
# End-of-turn is only used for UI hints - skip it
return None
async def _thread_item_to_input_item(
self,
item: ThreadItem,
is_last_message: bool = True,
) -> list[ChatMessage]:
"""Internal method to convert a single ThreadItem to ChatMessage(s).
Args:
item: The thread item to convert.
is_last_message: Whether this is the last item in the thread.
Returns:
A list of ChatMessage objects (may be empty).
"""
match item:
case UserMessageItem():
out = await self.user_message_to_input(item, is_last_message) or []
return out if isinstance(out, list) else [out]
case AssistantMessageItem():
out = await self.assistant_message_to_input(item) or []
return out if isinstance(out, list) else [out]
case ClientToolCallItem():
out = await self.client_tool_call_to_input(item) or []
return out if isinstance(out, list) else [out]
case EndOfTurnItem():
out = await self.end_of_turn_to_input(item) or []
return out if isinstance(out, list) else [out]
case WidgetItem():
out = self.widget_to_input(item) or []
return out if isinstance(out, list) else [out]
case WorkflowItem():
out = self.workflow_to_input(item) or []
return out if isinstance(out, list) else [out]
case TaskItem():
out = self.task_to_input(item) or []
return out if isinstance(out, list) else [out]
case HiddenContextItem():
out = self.hidden_context_to_input(item) or []
return out if isinstance(out, list) else [out]
case _:
assert_never(item)
async def to_agent_input(
self,
thread_items: Sequence[ThreadItem] | ThreadItem,
) -> list[ChatMessage]:
"""Convert ChatKit thread items to Agent Framework ChatMessages.
This is the main entry point for converting ChatKit thread items. It handles
all ThreadItem types (UserMessageItem, AssistantMessageItem, TaskItem, etc.)
and calls the appropriate conversion method for each.
Args:
thread_items: A single ThreadItem or a sequence of ThreadItems to convert.
Returns:
A list of ChatMessage objects that can be sent to an Agent Framework agent.
Examples:
.. code-block:: python
from agent_framework_chatkit import ThreadItemConverter
converter = ThreadItemConverter()
# Convert a single thread item
messages = await converter.to_agent_input(user_message_item)
# Convert multiple thread items
messages = await converter.to_agent_input([user_message_item, assistant_message_item, task_item])
# Use with agent
from agent_framework import ChatAgent
agent = ChatAgent(...)
response = await agent.run_stream(messages)
"""
thread_items = list(thread_items) if isinstance(thread_items, Sequence) else [thread_items]
output: list[ChatMessage] = []
for item in thread_items:
output.extend(
await self._thread_item_to_input_item(
item,
is_last_message=item is thread_items[-1],
)
)
return output
# Default converter instance
_DEFAULT_CONVERTER = ThreadItemConverter()
async def simple_to_agent_input(thread_items: Sequence[ThreadItem] | ThreadItem) -> list[ChatMessage]:
"""Helper function that uses the default ThreadItemConverter.
This function provides a quick way to get started with ChatKit integration
without needing to create a custom ThreadItemConverter instance.
Args:
thread_items: A single ThreadItem or a sequence of ThreadItems to convert.
Returns:
A list of ChatMessage objects that can be sent to an Agent Framework agent.
Examples:
.. code-block:: python
from agent_framework_chatkit import simple_to_agent_input
# Convert a single item
messages = await simple_to_agent_input(user_message_item)
# Convert multiple items
messages = await simple_to_agent_input([user_message_item, assistant_message_item, task_item])
"""
return await _DEFAULT_CONVERTER.to_agent_input(thread_items)
python/packages/chatkit/agent_framework_chatkit/_streaming.py
+104
View File
@@ -0,0 +1,104 @@
# Copyright (c) Microsoft. All rights reserved.
"""Streaming utilities for converting Agent Framework responses to ChatKit events."""
import uuid
from collections.abc import AsyncIterable, AsyncIterator, Callable
from datetime import datetime
from agent_framework import AgentRunResponseUpdate, TextContent
from chatkit.types import (
AssistantMessageContent,
AssistantMessageContentPartTextDelta,
AssistantMessageItem,
ThreadItemAddedEvent,
ThreadItemDoneEvent,
ThreadItemUpdated,
ThreadStreamEvent,
)
async def stream_agent_response(
response_stream: AsyncIterable[AgentRunResponseUpdate],
thread_id: str,
generate_id: Callable[[str], str] | None = None,
) -> AsyncIterator[ThreadStreamEvent]:
"""Convert a streamed AgentRunResponseUpdate from Agent Framework to ChatKit events.
This helper function takes a stream of AgentRunResponseUpdate objects from
a Microsoft Agent Framework agent and converts them to ChatKit ThreadStreamEvent
objects that can be consumed by the ChatKit UI.
The function supports real-time token-by-token streaming by emitting
ThreadItemUpdated events with AssistantMessageContentPartTextDelta for each
text chunk as it arrives from the agent.
Args:
response_stream: An async iterable of AgentRunResponseUpdate objects
from an Agent Framework agent.
thread_id: The ChatKit thread ID for the conversation.
generate_id: Optional function to generate IDs for ChatKit items.
If not provided, simple incremental IDs will be used.
Yields:
ThreadStreamEvent: ChatKit events representing the agent's response,
including incremental text deltas for streaming display.
"""
# Use provided ID generator or create default one
if generate_id is None:
def _default_id_generator(item_type: str) -> str:
return f"{item_type}_{uuid.uuid4().hex[:8]}"
message_id = _default_id_generator("msg")
else:
message_id = generate_id("msg")
# Track if we've started the message
message_started = False
accumulated_text = ""
content_index = 0
async for update in response_stream:
# Start the assistant message if not already started
if not message_started:
assistant_message = AssistantMessageItem(
id=message_id,
thread_id=thread_id,
type="assistant_message",
content=[],
created_at=datetime.now(),
)
yield ThreadItemAddedEvent(type="thread.item.added", item=assistant_message)
message_started = True
# Process the update content
if update.contents:
for content in update.contents:
# Handle text content - only TextContent has a text attribute
if isinstance(content, TextContent) and content.text is not None:
# Yield incremental text delta for streaming display
yield ThreadItemUpdated(
type="thread.item.updated",
item_id=message_id,
update=AssistantMessageContentPartTextDelta(
content_index=content_index,
delta=content.text,
),
)
accumulated_text += content.text
# Finalize the message
if message_started:
final_message = AssistantMessageItem(
id=message_id,
thread_id=thread_id,
type="assistant_message",
content=[AssistantMessageContent(type="output_text", text=accumulated_text, annotations=[])]
if accumulated_text
else [],
created_at=datetime.now(),
)
yield ThreadItemDoneEvent(type="thread.item.done", item=final_message)
python/packages/chatkit/agent_framework_chatkit/py.typed
View File
python/packages/chatkit/pyproject.toml
+89
View File
@@ -0,0 +1,89 @@
[project]
name = "agent-framework-chatkit"
description = "OpenAI ChatKit integration for Microsoft Agent Framework."
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251001"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true"
urls.issues = "https://github.com/microsoft/agent-framework/issues"
classifiers = [
"License :: OSI Approved :: MIT License",
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Typing :: Typed",
]
dependencies = [
"agent-framework-core",
"openai-chatkit>=1.1.0,<2.0.0",
]
[tool.uv]
prerelease = "if-necessary-or-explicit"
environments = [
"sys_platform == 'darwin'",
"sys_platform == 'linux'",
"sys_platform == 'win32'"
]
[tool.uv-dynamic-versioning]
fallback-version = "0.0.0"
[tool.pytest.ini_options]
testpaths = 'tests'
addopts = "-ra -q -r fEX"
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
filterwarnings = []
timeout = 120
[tool.ruff]
extend = "../../pyproject.toml"
[tool.ruff.lint]
ignore = ["RUF029"]
[tool.coverage.run]
omit = [
"**/__init__.py"
]
[tool.pyright]
extend = "../../pyproject.toml"
exclude = ['tests', 'chatkit-python', 'openai-chatkit-advanced-samples']
[tool.mypy]
plugins = ['pydantic.mypy']
strict = true
python_version = "3.10"
ignore_missing_imports = true
disallow_untyped_defs = true
no_implicit_optional = true
check_untyped_defs = true
warn_return_any = true
show_error_codes = true
warn_unused_ignores = false
disallow_incomplete_defs = true
disallow_untyped_decorators = true
[tool.bandit]
targets = ["agent_framework_chatkit"]
exclude_dirs = ["tests"]
[tool.poe]
executor.type = "uv"
include = "../../shared_tasks.toml"
[tool.poe.tasks]
mypy = "mypy --config-file $POE_ROOT/pyproject.toml agent_framework_chatkit"
test = "pytest --cov=agent_framework_chatkit --cov-report=term-missing:skip-covered tests"
[build-system]
requires = ["flit-core >= 3.11,<4.0"]
build-backend = "flit_core.buildapi"
python/packages/chatkit/tests/__init__.py
+1
View File
@@ -0,0 +1 @@
# Copyright (c) Microsoft. All rights reserved.
python/packages/chatkit/tests/test_converter.py
+426
View File
@@ -0,0 +1,426 @@
# Copyright (c) Microsoft. All rights reserved.
"""Tests for ChatKit to Agent Framework converter utilities."""
from unittest.mock import Mock
import pytest
from agent_framework import ChatMessage, Role, TextContent
from chatkit.types import UserMessageTextContent
from agent_framework_chatkit import ThreadItemConverter, simple_to_agent_input
class TestThreadItemConverter:
"""Tests for ThreadItemConverter class."""
@pytest.fixture
def converter(self):
"""Create a ThreadItemConverter instance for testing."""
return ThreadItemConverter()
async def test_to_agent_input_none(self, converter):
"""Test converting empty list returns empty list."""
result = await converter.to_agent_input([])
assert result == []
async def test_to_agent_input_with_text(self, converter):
"""Test converting user message with text content."""
from datetime import datetime
from chatkit.types import UserMessageItem
input_item = UserMessageItem(
id="msg_1",
thread_id="thread_1",
created_at=datetime.now(),
type="user_message",
content=[UserMessageTextContent(text="Hello, how can you help me?")],
attachments=[],
inference_options={},
)
result = await converter.to_agent_input(input_item)
assert len(result) == 1
assert isinstance(result[0], ChatMessage)
assert result[0].role == Role.USER
assert result[0].text == "Hello, how can you help me?"
async def test_to_agent_input_empty_text(self, converter):
"""Test converting user message with empty or whitespace-only text."""
from datetime import datetime
from chatkit.types import UserMessageItem
input_item = UserMessageItem(
id="msg_1",
thread_id="thread_1",
created_at=datetime.now(),
type="user_message",
content=[UserMessageTextContent(text=" ")],
attachments=[],
inference_options={},
)
result = await converter.to_agent_input(input_item)
assert result == []
async def test_to_agent_input_no_content(self, converter):
"""Test converting user message with no content."""
from datetime import datetime
from chatkit.types import UserMessageItem
input_item = UserMessageItem(
id="msg_1",
thread_id="thread_1",
created_at=datetime.now(),
type="user_message",
content=[],
attachments=[],
inference_options={},
)
result = await converter.to_agent_input(input_item)
assert result == []
async def test_to_agent_input_multiple_content_parts(self, converter):
"""Test converting user message with multiple text content parts."""
from datetime import datetime
from chatkit.types import UserMessageItem
input_item = UserMessageItem(
id="msg_1",
thread_id="thread_1",
created_at=datetime.now(),
type="user_message",
content=[
UserMessageTextContent(text="Hello "),
UserMessageTextContent(text="world!"),
],
attachments=[],
inference_options={},
)
result = await converter.to_agent_input(input_item)
assert len(result) == 1
assert result[0].text == "Hello world!"
def test_hidden_context_to_input(self, converter):
"""Test converting hidden context item to ChatMessage."""
hidden_item = Mock()
hidden_item.content = "This is hidden context information"
result = converter.hidden_context_to_input(hidden_item)
assert isinstance(result, ChatMessage)
assert result.role == Role.SYSTEM
assert result.text == "<HIDDEN_CONTEXT>This is hidden context information</HIDDEN_CONTEXT>"
def test_tag_to_message_content(self, converter):
"""Test converting tag to message content."""
from chatkit.types import UserMessageTagContent
tag = UserMessageTagContent(
type="input_tag",
id="tag_1",
text="john",
data={"name": "John Doe"},
interactive=False,
)
result = converter.tag_to_message_content(tag)
assert isinstance(result, TextContent)
# Since data is a dict, getattr won't work, so it will fall back to text
assert result.text == "<TAG>Name:john</TAG>"
def test_tag_to_message_content_no_name(self, converter):
"""Test converting tag with no name to message content."""
from chatkit.types import UserMessageTagContent
tag = UserMessageTagContent(
type="input_tag",
id="tag_2",
text="jane",
data={},
interactive=False,
)
result = converter.tag_to_message_content(tag)
assert isinstance(result, TextContent)
assert result.text == "<TAG>Name:jane</TAG>"
async def test_attachment_to_message_content_file_without_fetcher(self, converter):
"""Test that FileAttachment without data fetcher returns None."""
from chatkit.types import FileAttachment
attachment = FileAttachment(
id="file_123",
name="document.pdf",
mime_type="application/pdf",
type="file",
)
result = await converter.attachment_to_message_content(attachment)
assert result is None
async def test_attachment_to_message_content_image_with_preview_url(self, converter):
"""Test that ImageAttachment with preview_url creates UriContent."""
from agent_framework import UriContent
from chatkit.types import ImageAttachment
attachment = ImageAttachment(
id="img_123",
name="photo.jpg",
mime_type="image/jpeg",
type="image",
preview_url="https://example.com/photo.jpg",
)
result = await converter.attachment_to_message_content(attachment)
assert isinstance(result, UriContent)
assert result.uri == "https://example.com/photo.jpg"
assert result.media_type == "image/jpeg"
async def test_attachment_to_message_content_with_data_fetcher(self):
"""Test attachment conversion with data fetcher."""
from agent_framework import DataContent
from chatkit.types import FileAttachment
# Mock data fetcher
async def fetch_data(attachment_id: str) -> bytes:
return b"file content data"
converter = ThreadItemConverter(attachment_data_fetcher=fetch_data)
attachment = FileAttachment(
id="file_123",
name="document.pdf",
mime_type="application/pdf",
type="file",
)
result = await converter.attachment_to_message_content(attachment)
assert isinstance(result, DataContent)
assert result.media_type == "application/pdf"
async def test_to_agent_input_with_image_attachment(self):
"""Test converting user message with text and image attachment."""
from datetime import datetime
from agent_framework import UriContent
from chatkit.types import ImageAttachment, UserMessageItem
attachment = ImageAttachment(
id="img_123",
name="photo.jpg",
mime_type="image/jpeg",
type="image",
preview_url="https://example.com/photo.jpg",
)
input_item = UserMessageItem(
id="msg_1",
thread_id="thread_1",
created_at=datetime.now(),
type="user_message",
content=[UserMessageTextContent(text="Check out this photo!")],
attachments=[attachment],
inference_options={},
)
converter = ThreadItemConverter()
result = await converter.to_agent_input(input_item)
assert len(result) == 1
message = result[0]
assert message.role == Role.USER
assert len(message.contents) == 2
# First content should be text
assert isinstance(message.contents[0], TextContent)
assert message.contents[0].text == "Check out this photo!"
# Second content should be UriContent for the image
assert isinstance(message.contents[1], UriContent)
assert message.contents[1].uri == "https://example.com/photo.jpg"
assert message.contents[1].media_type == "image/jpeg"
async def test_to_agent_input_with_file_attachment_and_fetcher(self):
"""Test converting user message with file attachment using data fetcher."""
from datetime import datetime
from agent_framework import DataContent
from chatkit.types import FileAttachment, UserMessageItem
attachment = FileAttachment(
id="file_123",
name="report.pdf",
mime_type="application/pdf",
type="file",
)
input_item = UserMessageItem(
id="msg_1",
thread_id="thread_1",
created_at=datetime.now(),
type="user_message",
content=[UserMessageTextContent(text="Here's the document")],
attachments=[attachment],
inference_options={},
)
# Create converter with data fetcher
async def fetch_data(attachment_id: str) -> bytes:
return b"PDF content data"
converter = ThreadItemConverter(attachment_data_fetcher=fetch_data)
result = await converter.to_agent_input(input_item)
assert len(result) == 1
message = result[0]
assert len(message.contents) == 2
# First content should be text
assert isinstance(message.contents[0], TextContent)
# Second content should be DataContent for the file
assert isinstance(message.contents[1], DataContent)
assert message.contents[1].media_type == "application/pdf"
def test_task_to_input(self, converter):
"""Test converting TaskItem to ChatMessage."""
from datetime import datetime
from chatkit.types import CustomTask, TaskItem
task_item = TaskItem(
id="task_1",
thread_id="thread_1",
created_at=datetime.now(),
type="task",
task=CustomTask(type="custom", title="Analysis", content="Analyzed the data"),
)
result = converter.task_to_input(task_item)
assert isinstance(result, ChatMessage)
assert result.role == Role.USER
assert "Analysis: Analyzed the data" in result.text
assert "<Task>" in result.text
def test_task_to_input_no_custom_task(self, converter):
"""Test that non-custom tasks return None."""
from datetime import datetime
from chatkit.types import TaskItem, ThoughtTask
task_item = TaskItem(
id="task_1",
thread_id="thread_1",
created_at=datetime.now(),
type="task",
task=ThoughtTask(type="thought", title="Think", content="Thinking..."),
)
result = converter.task_to_input(task_item)
assert result is None
def test_workflow_to_input(self, converter):
"""Test converting WorkflowItem to ChatMessages."""
from datetime import datetime
from chatkit.types import CustomTask, Workflow, WorkflowItem
workflow_item = WorkflowItem(
id="wf_1",
thread_id="thread_1",
created_at=datetime.now(),
type="workflow",
workflow=Workflow(
type="custom",
tasks=[
CustomTask(type="custom", title="Step 1", content="First step"),
CustomTask(type="custom", title="Step 2", content="Second step"),
],
),
)
result = converter.workflow_to_input(workflow_item)
assert isinstance(result, list)
assert len(result) == 2
assert all(isinstance(msg, ChatMessage) for msg in result)
assert "Step 1: First step" in result[0].text
assert "Step 2: Second step" in result[1].text
def test_workflow_to_input_empty(self, converter):
"""Test that workflows with no custom tasks return None."""
from datetime import datetime
from chatkit.types import Workflow, WorkflowItem
workflow_item = WorkflowItem(
id="wf_1",
thread_id="thread_1",
created_at=datetime.now(),
type="workflow",
workflow=Workflow(type="custom", tasks=[]),
)
result = converter.workflow_to_input(workflow_item)
assert result is None
def test_widget_to_input(self, converter):
"""Test converting WidgetItem to ChatMessage."""
from datetime import datetime
from chatkit.types import WidgetItem
from chatkit.widgets import Card, Text
widget_item = WidgetItem(
id="widget_1",
thread_id="thread_1",
created_at=datetime.now(),
type="widget",
widget=Card(key="card1", children=[Text(value="Hello")]),
)
result = converter.widget_to_input(widget_item)
assert isinstance(result, ChatMessage)
assert result.role == Role.USER
assert "widget_1" in result.text
assert "graphical UI widget" in result.text
class TestSimpleToAgentInput:
"""Tests for simple_to_agent_input helper function."""
async def test_simple_to_agent_input_empty_list(self):
"""Test simple conversion with empty list."""
result = await simple_to_agent_input([])
assert result == []
async def test_simple_to_agent_input_with_text(self):
"""Test simple conversion with text content."""
from datetime import datetime
from chatkit.types import UserMessageItem
input_item = UserMessageItem(
id="msg_1",
thread_id="thread_1",
created_at=datetime.now(),
type="user_message",
content=[UserMessageTextContent(text="Test message")],
attachments=[],
inference_options={},
)
result = await simple_to_agent_input(input_item)
assert len(result) == 1
assert isinstance(result[0], ChatMessage)
assert result[0].role == Role.USER
assert result[0].text == "Test message"
python/packages/chatkit/tests/test_streaming.py
+142
View File
@@ -0,0 +1,142 @@
# Copyright (c) Microsoft. All rights reserved.
"""Tests for Agent Framework to ChatKit streaming utilities."""
from unittest.mock import Mock
from agent_framework import AgentRunResponseUpdate, Role, TextContent
from chatkit.types import (
ThreadItemAddedEvent,
ThreadItemDoneEvent,
ThreadItemUpdated,
)
from agent_framework_chatkit import stream_agent_response
class TestStreamAgentResponse:
"""Tests for stream_agent_response function."""
async def test_stream_empty_response(self):
"""Test streaming empty response."""
async def empty_stream():
return
yield # Make it a generator
events = []
async for event in stream_agent_response(empty_stream(), thread_id="test_thread"):
events.append(event)
assert len(events) == 0
async def test_stream_single_text_update(self):
"""Test streaming single text update."""
async def single_update_stream():
yield AgentRunResponseUpdate(role=Role.ASSISTANT, contents=[TextContent(text="Hello world")])
events = []
async for event in stream_agent_response(single_update_stream(), thread_id="test_thread"):
events.append(event)
# Should have: item_added, item_updated (delta), item_done
assert len(events) == 3
# Check event types
assert isinstance(events[0], ThreadItemAddedEvent)
assert isinstance(events[1], ThreadItemUpdated)
assert isinstance(events[2], ThreadItemDoneEvent)
# Check delta event
assert events[1].update.delta == "Hello world"
# Check final message content
assert len(events[2].item.content) == 1
assert events[2].item.content[0].text == "Hello world"
async def test_stream_multiple_text_updates(self):
"""Test streaming multiple text updates."""
async def multiple_updates_stream():
yield AgentRunResponseUpdate(role=Role.ASSISTANT, contents=[TextContent(text="Hello ")])
yield AgentRunResponseUpdate(role=Role.ASSISTANT, contents=[TextContent(text="world!")])
events = []
async for event in stream_agent_response(multiple_updates_stream(), thread_id="test_thread"):
events.append(event)
# Should have: item_added, item_updated (delta 1), item_updated (delta 2), item_done
assert len(events) == 4
# Check event types
assert isinstance(events[0], ThreadItemAddedEvent)
assert isinstance(events[1], ThreadItemUpdated)
assert isinstance(events[2], ThreadItemUpdated)
assert isinstance(events[3], ThreadItemDoneEvent)
# Check delta events
assert events[1].update.delta == "Hello "
assert events[2].update.delta == "world!"
# Check final accumulated text
final_message_event = events[-1]
assert isinstance(final_message_event, ThreadItemDoneEvent)
assert final_message_event.item.content[0].text == "Hello world!"
async def test_stream_with_custom_id_generator(self):
"""Test streaming with custom ID generator."""
def custom_id_generator(item_type: str) -> str:
return f"custom_{item_type}_123"
async def single_update_stream():
yield AgentRunResponseUpdate(role=Role.ASSISTANT, contents=[TextContent(text="Test")])
events = []
async for event in stream_agent_response(
single_update_stream(), thread_id="test_thread", generate_id=custom_id_generator
):
events.append(event)
# Check that custom IDs are used
message_added_event = events[0]
assert message_added_event.item.id == "custom_msg_123"
async def test_stream_empty_content_updates(self):
"""Test streaming updates with empty content."""
async def empty_content_stream():
yield AgentRunResponseUpdate(role=Role.ASSISTANT, contents=[])
yield AgentRunResponseUpdate(role=Role.ASSISTANT, contents=None)
events = []
async for event in stream_agent_response(empty_content_stream(), thread_id="test_thread"):
events.append(event)
# Should have item_added and item_done
assert len(events) == 2
assert isinstance(events[0], ThreadItemAddedEvent)
assert isinstance(events[1], ThreadItemDoneEvent)
# Final message should have empty content
assert len(events[1].item.content) == 0
async def test_stream_non_text_content(self):
"""Test streaming updates with non-text content."""
# Mock a content object without text attribute
non_text_content = Mock()
# Don't set text attribute
del non_text_content.text
async def non_text_stream():
yield AgentRunResponseUpdate(role=Role.ASSISTANT, contents=[non_text_content])
events = []
async for event in stream_agent_response(non_text_stream(), thread_id="test_thread"):
events.append(event)
# Should have item_added and item_done, but no content since no text
assert len(events) == 2
assert isinstance(events[0], ThreadItemAddedEvent)
assert isinstance(events[1], ThreadItemDoneEvent)
python/packages/core/agent_framework/chatkit/__init__.py
+23
View File
@@ -0,0 +1,23 @@
# Copyright (c) Microsoft. All rights reserved.
import importlib
from typing import Any
PACKAGE_NAME = "agent_framework_chatkit"
PACKAGE_EXTRA = "chatkit"
_IMPORTS = ["__version__", "ThreadItemConverter", "simple_to_agent_input", "stream_agent_response"]
def __getattr__(name: str) -> Any:
if name in _IMPORTS:
try:
return getattr(importlib.import_module(PACKAGE_NAME), name)
except ModuleNotFoundError as exc:
raise ModuleNotFoundError(
f"The '{PACKAGE_EXTRA}' extra is not installed, please do `pip install agent-framework-{PACKAGE_EXTRA}`"
) from exc
raise AttributeError(f"Module {PACKAGE_NAME} has no attribute {name}.")
def __dir__() -> list[str]:
return _IMPORTS
python/packages/core/agent_framework/chatkit/__init__.pyi
+10
View File
@@ -0,0 +1,10 @@
# Copyright (c) Microsoft. All rights reserved.
from agent_framework_chatkit import (
ThreadItemConverter,
__version__,
simple_to_agent_input,
stream_agent_response,
)
__all__ = ["ThreadItemConverter", "__version__", "simple_to_agent_input", "stream_agent_response"]
python/pyproject.toml
+3
View File
@@ -26,6 +26,7 @@ dependencies = [
"agent-framework-a2a",
"agent-framework-anthropic",
"agent-framework-azure-ai",
"agent-framework-chatkit",
"agent-framework-copilotstudio",
"agent-framework-devui",
"agent-framework-lab",
@@ -89,6 +90,7 @@ agent-framework = { workspace = true }
agent-framework-core = { workspace = true }
agent-framework-a2a = { workspace = true }
agent-framework-azure-ai = { workspace = true }
agent-framework-chatkit = { workspace = true }
agent-framework-copilotstudio = { workspace = true }
agent-framework-lab = { workspace = true }
agent-framework-mem0 = { workspace = true }
@@ -240,6 +242,7 @@ pytest --import-mode=importlib
--cov=agent_framework
--cov=agent_framework_a2a
--cov=agent_framework_azure_ai
--cov=agent_framework_chatkit
--cov=agent_framework_copilotstudio
--cov=agent_framework_mem0
--cov=agent_framework_redis
python/samples/demos/chatkit-integration/.gitignore
+4
View File
@@ -0,0 +1,4 @@
*.db
*.db-shm
*.db-wal
uploads/
python/samples/demos/chatkit-integration/README.md
+268
View File
@@ -0,0 +1,268 @@
# ChatKit Integration Sample with Weather Agent and Image Analysis
This sample demonstrates how to integrate Microsoft Agent Framework with OpenAI ChatKit. It provides a complete implementation of a weather assistant with interactive widget visualization, image analysis, and file upload support.
**Features:**
- Weather information with interactive widgets
- Image analysis using vision models
- Current time queries
- File upload with attachment storage
- Chat interface with streaming responses
- City selector widget with one-click weather
## Architecture
```mermaid
graph TB
subgraph Frontend["React Frontend (ChatKit UI)"]
UI[ChatKit Components]
Upload[File Upload]
end
subgraph Backend["FastAPI Server"]
FastAPI[FastAPI Endpoints]
subgraph ChatKit["WeatherChatKitServer"]
Respond[respond method]
Action[action method]
end
subgraph Stores["Data & Storage Layer"]
SQLite[SQLiteStore<br/>Store Protocol]
AttStore[FileBasedAttachmentStore<br/>AttachmentStore Protocol]
DB[(SQLite DB<br/>chatkit_demo.db)]
Files[/uploads directory/]
end
subgraph Integration["Agent Framework Integration"]
Converter[ThreadItemConverter]
Streamer[stream_agent_response]
Agent[ChatAgent]
end
Widgets[Widget Rendering<br/>render_weather_widget<br/>render_city_selector_widget]
end
subgraph Azure["Azure AI"]
Foundry[GPT-5<br/>with Vision]
end
UI -->|HTTP POST /chatkit| FastAPI
Upload -->|HTTP POST /upload/id| FastAPI
FastAPI --> ChatKit
ChatKit -->|save/load threads| SQLite
ChatKit -->|save/load attachments| AttStore
ChatKit -->|convert messages| Converter
SQLite -.->|persist| DB
AttStore -.->|save files| Files
AttStore -.->|save metadata| SQLite
Converter -->|ChatMessage array| Agent
Agent -->|AgentRunResponseUpdate| Streamer
Streamer -->|ThreadStreamEvent| ChatKit
ChatKit --> Widgets
Widgets -->|WidgetItem| ChatKit
Agent <-->|Chat Completions API| Foundry
ChatKit -->|ThreadStreamEvent| FastAPI
FastAPI -->|SSE Stream| UI
style ChatKit fill:#e1f5ff
style Stores fill:#fff4e1
style Integration fill:#f0e1ff
style Azure fill:#e1ffe1
```
### Server Implementation
The sample implements a ChatKit server using the `ChatKitServer` base class from the `chatkit` package:
**Core Components:**
- **`WeatherChatKitServer`**: Custom ChatKit server implementation that:
- Extends `ChatKitServer[dict[str, Any]]`
- Uses Agent Framework's `ChatAgent` with Azure OpenAI
- Converts ChatKit messages to Agent Framework format using `ThreadItemConverter`
- Streams responses back to ChatKit using `stream_agent_response`
- Creates and streams interactive widgets after agent responses
- **`SQLiteStore`**: Data persistence layer that:
- Implements the `Store[dict[str, Any]]` protocol from ChatKit
- Persists threads, messages, and attachment metadata in SQLite
- Provides thread management and item history
- Stores attachment metadata for the upload lifecycle
- **`FileBasedAttachmentStore`**: File storage implementation that:
- Implements the `AttachmentStore[dict[str, Any]]` protocol from ChatKit
- Stores uploaded files on the local filesystem (in `./uploads` directory)
- Generates upload URLs for two-phase file upload
- Saves attachment metadata to the data store for upload tracking
- Provides preview URLs for images
**Key Integration Points:**
```python
# Converting ChatKit messages to Agent Framework
converter = ThreadItemConverter(
attachment_data_fetcher=self._fetch_attachment_data
)
agent_messages = await converter.to_agent_input(user_message_item)
# Running agent and streaming back to ChatKit
async for event in stream_agent_response(
self.weather_agent.run_stream(agent_messages),
thread_id=thread.id,
):
yield event
# Streaming widgets
widget = render_weather_widget(weather_data)
async for event in stream_widget(thread_id=thread.id, widget=widget):
yield event
```
## Installation and Setup
### Prerequisites
- Python 3.10+
- Node.js 18.18+ and npm 9+
- Azure OpenAI service configured
- Azure CLI for authentication (`az login`)
### Backend Setup
1. **Install Python packages:**
```bash
cd python/samples/demos/chatkit-integration
pip install agent-framework-chatkit fastapi uvicorn azure-identity
```
2. **Configure Azure OpenAI:**
```bash
export AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com/"
export AZURE_OPENAI_API_VERSION="2024-06-01"
export AZURE_OPENAI_CHAT_DEPLOYMENT_NAME="gpt-4o"
```
3. **Authenticate with Azure:**
```bash
az login
```
### Frontend Setup
Install the Node.js dependencies:
```bash
cd frontend
npm install
```
## How to Run
### Start the Backend Server
From the `chatkit-integration` directory:
```bash
python app.py
```
Or with auto-reload for development:
```bash
uvicorn app:app --host 127.0.0.1 --port 8001 --reload
```
The backend will start on `http://localhost:8001`
### Start the Frontend Development Server
In a new terminal, from the `frontend` directory:
```bash
npm run dev
```
The frontend will start on `http://localhost:5171`
### Access the Application
Open your browser and navigate to:
```
http://localhost:5171
```
You can now:
- Ask about weather in any location (weather widgets display automatically)
- Upload images for analysis using the attachment button
- Get the current time
- Ask to see available cities and click city buttons for instant weather
### Project Structure
```
chatkit-integration/
├── app.py # FastAPI backend with ChatKitServer implementation
├── store.py # SQLiteStore implementation
├── attachment_store.py # FileBasedAttachmentStore implementation
├── weather_widget.py # Widget rendering functions
├── chatkit_demo.db # SQLite database (auto-created)
├── uploads/ # Uploaded files directory (auto-created)
└── frontend/
├── package.json
├── vite.config.ts
├── index.html
└── src/
├── main.tsx
└── App.tsx # ChatKit UI integration
```
### Configuration
You can customize the application by editing constants at the top of `app.py`:
```python
# Server configuration
SERVER_HOST = "127.0.0.1" # Bind to localhost only for security (local dev)
SERVER_PORT = 8001
SERVER_BASE_URL = f"http://localhost:{SERVER_PORT}"
# Database configuration
DATABASE_PATH = "chatkit_demo.db"
# File storage configuration
UPLOADS_DIRECTORY = "./uploads"
# User context
DEFAULT_USER_ID = "demo_user"
```
### Sample Conversations
Try these example queries:
- "What's the weather like in Tokyo?"
- "Show me available cities" (displays interactive city selector)
- "What's the current time?"
- Upload an image and ask "What do you see in this image?"
## Learn More
- [Agent Framework Documentation](https://aka.ms/agent-framework)
- [ChatKit Documentation](https://platform.openai.com/docs/guides/chatkit)
- [Azure OpenAI Documentation](https://learn.microsoft.com/en-us/azure/ai-foundry/)
python/samples/demos/chatkit-integration/__init__.py
+1
View File
@@ -0,0 +1 @@
# Copyright (c) Microsoft. All rights reserved.
python/samples/demos/chatkit-integration/app.py
+538
View File
@@ -0,0 +1,538 @@
# Copyright (c) Microsoft. All rights reserved.
"""
ChatKit Integration Sample with Weather Agent and Image Analysis
This sample demonstrates how to integrate Microsoft Agent Framework with OpenAI ChatKit
using a weather tool with widget visualization, image analysis, and Azure OpenAI. It shows
a complete ChatKit server implementation using Agent Framework agents with proper FastAPI
setup, interactive weather widgets, and vision capabilities for analyzing uploaded images.
"""
import logging
from collections.abc import AsyncIterator, Callable
from datetime import datetime, timezone
from random import randint
from typing import Annotated, Any
import uvicorn
from azure.identity import AzureCliCredential
from fastapi import FastAPI, File, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse, Response, StreamingResponse
from pydantic import Field
# ============================================================================
# Configuration Constants
# ============================================================================
# Server configuration
SERVER_HOST = "127.0.0.1" # Bind to localhost only for security (local dev)
SERVER_PORT = 8001
SERVER_BASE_URL = f"http://localhost:{SERVER_PORT}"
# Database configuration
DATABASE_PATH = "chatkit_demo.db"
# File storage configuration
UPLOADS_DIRECTORY = "./uploads"
# User context
DEFAULT_USER_ID = "demo_user"
# Logging configuration
LOG_LEVEL = logging.INFO
LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
# ============================================================================
# Logging Setup
# ============================================================================
logging.basicConfig(
level=LOG_LEVEL,
format=LOG_FORMAT,
datefmt=LOG_DATE_FORMAT,
)
logger = logging.getLogger(__name__)
# Agent Framework imports
from agent_framework import AgentRunResponseUpdate, ChatAgent, ChatMessage, FunctionResultContent, Role
from agent_framework.azure import AzureOpenAIChatClient
# Agent Framework ChatKit integration
from agent_framework_chatkit import ThreadItemConverter, stream_agent_response
# Local imports
from attachment_store import FileBasedAttachmentStore
# ChatKit imports
from chatkit.actions import Action
from chatkit.server import ChatKitServer
from chatkit.store import StoreItemType, default_generate_id
from chatkit.types import (
ThreadItemDoneEvent,
ThreadMetadata,
ThreadStreamEvent,
UserMessageItem,
WidgetItem,
)
from chatkit.widgets import WidgetRoot
from store import SQLiteStore
from weather_widget import (
WeatherData,
city_selector_copy_text,
render_city_selector_widget,
render_weather_widget,
weather_widget_copy_text,
)
class WeatherResponse(str):
"""A string response that also carries WeatherData for widget creation."""
def __new__(cls, text: str, weather_data: WeatherData):
instance = super().__new__(cls, text)
instance.weather_data = weather_data # type: ignore
return instance
async def stream_widget(
thread_id: str,
widget: WidgetRoot,
copy_text: str | None = None,
generate_id: Callable[[StoreItemType], str] = default_generate_id,
) -> AsyncIterator[ThreadStreamEvent]:
"""Stream a ChatKit widget as a ThreadStreamEvent.
This helper function creates a ChatKit widget item and yields it as a
ThreadItemDoneEvent that can be consumed by the ChatKit UI.
Args:
thread_id: The ChatKit thread ID for the conversation.
widget: The ChatKit widget to display.
copy_text: Optional text representation of the widget for copy/paste.
generate_id: Optional function to generate IDs for ChatKit items.
Yields:
ThreadStreamEvent: ChatKit event containing the widget.
"""
item_id = generate_id("message")
widget_item = WidgetItem(
id=item_id,
thread_id=thread_id,
created_at=datetime.now(),
widget=widget,
copy_text=copy_text,
)
yield ThreadItemDoneEvent(type="thread.item.done", item=widget_item)
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location.
Returns a string description with embedded WeatherData for widget creation.
"""
logger.info(f"Fetching weather for location: {location}")
conditions = ["sunny", "cloudy", "rainy", "stormy", "snowy", "foggy"]
temperature = randint(-5, 35)
condition = conditions[randint(0, len(conditions) - 1)]
# Add some realistic details
humidity = randint(30, 90)
wind_speed = randint(5, 25)
weather_data = WeatherData(
location=location,
condition=condition,
temperature=temperature,
humidity=humidity,
wind_speed=wind_speed,
)
logger.debug(f"Weather data generated: {condition}, {temperature}°C, {humidity}% humidity, {wind_speed} km/h wind")
# Return a WeatherResponse that is both a string (for the LLM) and carries structured data
text = (
f"Weather in {location}:\n"
f"• Condition: {condition.title()}\n"
f"• Temperature: {temperature}°C\n"
f"• Humidity: {humidity}%\n"
f"• Wind: {wind_speed} km/h"
)
return WeatherResponse(text, weather_data)
def get_time() -> str:
"""Get the current UTC time."""
current_time = datetime.now(timezone.utc)
logger.info("Getting current UTC time")
return f"Current UTC time: {current_time.strftime('%Y-%m-%d %H:%M:%S')} UTC"
def show_city_selector() -> str:
"""Show an interactive city selector widget to the user.
This function triggers the display of a widget that allows users
to select from popular cities to get weather information.
Returns a special marker string that will be detected to show the widget.
"""
logger.info("Activating city selector widget")
return "__SHOW_CITY_SELECTOR__"
class WeatherChatKitServer(ChatKitServer[dict[str, Any]]):
"""ChatKit server implementation using Agent Framework.
This server integrates Agent Framework agents with ChatKit's server protocol,
providing weather information with interactive widgets and time queries through Azure OpenAI.
"""
def __init__(self, data_store: SQLiteStore, attachment_store: FileBasedAttachmentStore):
super().__init__(data_store, attachment_store)
logger.info("Initializing WeatherChatKitServer")
# Create Agent Framework agent with Azure OpenAI
# For authentication, run `az login` command in terminal
try:
self.weather_agent = ChatAgent(
chat_client=AzureOpenAIChatClient(credential=AzureCliCredential()),
instructions=(
"You are a helpful weather assistant with image analysis capabilities. "
"You can provide weather information for any location, tell the current time, "
"and analyze images that users upload. Be friendly and informative in your responses.\n\n"
"If a user asks to see a list of cities or wants to choose from available cities, "
"use the show_city_selector tool to display an interactive city selector.\n\n"
"When users upload images, you will automatically receive them and can analyze their content. "
"Describe what you see in detail and be helpful in answering questions about the images."
),
tools=[get_weather, get_time, show_city_selector],
)
logger.info("Weather agent initialized successfully with Azure OpenAI")
except Exception as e:
logger.error(f"Failed to initialize weather agent: {e}")
raise
# Create ThreadItemConverter with attachment data fetcher
self.converter = ThreadItemConverter(
attachment_data_fetcher=self._fetch_attachment_data,
)
logger.info("WeatherChatKitServer initialized")
async def _fetch_attachment_data(self, attachment_id: str) -> bytes:
"""Fetch attachment binary data for the converter.
Args:
attachment_id: The ID of the attachment to fetch.
Returns:
The binary data of the attachment.
"""
return await attachment_store.read_attachment_bytes(attachment_id)
async def respond(
self,
thread: ThreadMetadata,
input_user_message: UserMessageItem | None,
context: dict[str, Any],
) -> AsyncIterator[ThreadStreamEvent]:
"""Handle incoming user messages and generate responses.
This method converts ChatKit messages to Agent Framework format using ThreadItemConverter,
runs the agent, converts the response back to ChatKit events using stream_agent_response,
and creates interactive weather widgets when weather data is queried.
"""
from agent_framework import FunctionResultContent
if input_user_message is None:
logger.debug("Received None user message, skipping")
return
logger.info(f"Processing message for thread: {thread.id}")
try:
# Track weather data and city selector flag for this request
weather_data: WeatherData | None = None
show_city_selector = False
# Convert ChatKit user message to Agent Framework ChatMessage using ThreadItemConverter
agent_messages = await self.converter.to_agent_input(input_user_message)
if not agent_messages:
logger.warning("No messages after conversion")
return
logger.info(f"Running agent with {len(agent_messages)} message(s)")
# Run the Agent Framework agent with streaming
agent_stream = self.weather_agent.run_stream(agent_messages)
# Create an intercepting stream that extracts function results while passing through updates
async def intercept_stream() -> AsyncIterator[AgentRunResponseUpdate]:
nonlocal weather_data, show_city_selector
async for update in agent_stream:
# Check for function results in the update
if update.contents:
for content in update.contents:
if isinstance(content, FunctionResultContent):
result = content.result
# Check if it's a WeatherResponse (string subclass with weather_data attribute)
if isinstance(result, str) and hasattr(result, "weather_data"):
extracted_data = getattr(result, "weather_data", None)
if isinstance(extracted_data, WeatherData):
weather_data = extracted_data
logger.info(f"Weather data extracted: {weather_data.location}")
# Check if it's the city selector marker
elif isinstance(result, str) and result == "__SHOW_CITY_SELECTOR__":
show_city_selector = True
logger.info("City selector flag detected")
yield update
# Stream updates as ChatKit events with interception
async for event in stream_agent_response(
intercept_stream(),
thread_id=thread.id,
):
yield event
# If weather data was collected during the tool call, create a widget
if weather_data is not None and isinstance(weather_data, WeatherData):
logger.info(f"Creating weather widget for location: {weather_data.location}")
# Create weather widget
widget = render_weather_widget(weather_data)
copy_text = weather_widget_copy_text(weather_data)
# Stream the widget
async for widget_event in stream_widget(thread_id=thread.id, widget=widget, copy_text=copy_text):
yield widget_event
logger.debug("Weather widget streamed successfully")
# If city selector should be shown, create and stream that widget
if show_city_selector:
logger.info("Creating city selector widget")
# Create city selector widget
selector_widget = render_city_selector_widget()
selector_copy_text = city_selector_copy_text()
# Stream the widget
async for widget_event in stream_widget(
thread_id=thread.id, widget=selector_widget, copy_text=selector_copy_text
):
yield widget_event
logger.debug("City selector widget streamed successfully")
logger.info(f"Completed processing message for thread: {thread.id}")
except Exception as e:
logger.error(f"Error processing message for thread {thread.id}: {e}", exc_info=True)
async def action(
self,
thread: ThreadMetadata,
action: Action[str, Any],
sender: WidgetItem | None,
context: dict[str, Any],
) -> AsyncIterator[ThreadStreamEvent]:
"""Handle widget actions from the frontend.
This method processes actions triggered by interactive widgets,
such as city selection from the city selector widget.
"""
logger.info(f"Received action: {action.type} for thread: {thread.id}")
if action.type == "city_selected":
# Extract city information from the action payload
city_label = action.payload.get("city_label", "Unknown")
logger.info(f"City selected: {city_label}")
logger.debug(f"Action payload: {action.payload}")
# Track weather data for this request
weather_data: WeatherData | None = None
# Create an agent message asking about the weather
agent_messages = [ChatMessage(role=Role.USER, text=f"What's the weather in {city_label}?")]
logger.debug(f"Processing weather query: {agent_messages[0].text}")
# Run the Agent Framework agent with streaming
agent_stream = self.weather_agent.run_stream(agent_messages)
# Create an intercepting stream that extracts function results while passing through updates
async def intercept_stream() -> AsyncIterator[AgentRunResponseUpdate]:
nonlocal weather_data
async for update in agent_stream:
# Check for function results in the update
if update.contents:
for content in update.contents:
if isinstance(content, FunctionResultContent):
result = content.result
# Check if it's a WeatherResponse (string subclass with weather_data attribute)
if isinstance(result, str) and hasattr(result, "weather_data"):
extracted_data = getattr(result, "weather_data", None)
if isinstance(extracted_data, WeatherData):
weather_data = extracted_data
logger.info(f"Weather data extracted: {weather_data.location}")
yield update
# Stream updates as ChatKit events with interception
async for event in stream_agent_response(
intercept_stream(),
thread_id=thread.id,
):
yield event
# If weather data was collected during the tool call, create a widget
if weather_data is not None and isinstance(weather_data, WeatherData):
logger.info(f"Creating weather widget for: {weather_data.location}")
# Create weather widget
widget = render_weather_widget(weather_data)
copy_text = weather_widget_copy_text(weather_data)
# Stream the widget
async for widget_event in stream_widget(thread_id=thread.id, widget=widget, copy_text=copy_text):
yield widget_event
logger.debug("Weather widget created successfully from action")
else:
logger.warning("No weather data available to create widget after action")
# FastAPI application setup
app = FastAPI(
title="ChatKit Weather & Vision Agent",
description="Weather and image analysis assistant powered by Agent Framework and Azure OpenAI",
version="1.0.0",
)
# Add CORS middleware to allow frontend connections
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # In production, specify exact origins
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Initialize data store and ChatKit server
logger.info("Initializing application components")
data_store = SQLiteStore(db_path=DATABASE_PATH)
attachment_store = FileBasedAttachmentStore(
uploads_dir=UPLOADS_DIRECTORY,
base_url=SERVER_BASE_URL,
data_store=data_store,
)
chatkit_server = WeatherChatKitServer(data_store, attachment_store)
logger.info("Application initialization complete")
@app.post("/chatkit")
async def chatkit_endpoint(request: Request):
"""Main ChatKit endpoint that handles all ChatKit requests.
This endpoint follows the ChatKit server protocol and handles both
streaming and non-streaming responses.
"""
logger.debug(f"Received ChatKit request from {request.client}")
request_body = await request.body()
# Create context following the working examples pattern
context = {"request": request}
try:
# Process the request using ChatKit server
result = await chatkit_server.process(request_body, context)
# Return appropriate response type
if hasattr(result, "__aiter__"): # StreamingResult
logger.debug("Returning streaming response")
return StreamingResponse(result, media_type="text/event-stream") # type: ignore[arg-type]
# NonStreamingResult
logger.debug("Returning non-streaming response")
return Response(content=result.json, media_type="application/json") # type: ignore[union-attr]
except Exception as e:
logger.error(f"Error processing ChatKit request: {e}", exc_info=True)
raise
@app.post("/upload/{attachment_id}")
async def upload_file(attachment_id: str, file: UploadFile = File(...)):
"""Handle file upload for two-phase upload.
The client POSTs the file bytes here after creating the attachment
via the ChatKit attachments.create endpoint.
"""
logger.info(f"Receiving file upload for attachment: {attachment_id}")
try:
# Read file contents
contents = await file.read()
# Save to disk
file_path = attachment_store.get_file_path(attachment_id)
file_path.write_bytes(contents)
logger.info(f"Saved {len(contents)} bytes to {file_path}")
# Load the attachment metadata from the data store
attachment = await data_store.load_attachment(attachment_id, {"user_id": DEFAULT_USER_ID})
# Clear the upload_url since upload is complete
attachment.upload_url = None
# Save the updated attachment back to the store
await data_store.save_attachment(attachment, {"user_id": DEFAULT_USER_ID})
# Return the attachment metadata as JSON
return JSONResponse(content=attachment.model_dump(mode="json"))
except Exception as e:
logger.error(f"Error uploading file for attachment {attachment_id}: {e}", exc_info=True)
return JSONResponse(status_code=500, content={"error": f"Failed to upload file: {str(e)}"})
@app.get("/preview/{attachment_id}")
async def preview_image(attachment_id: str):
"""Serve image preview/thumbnail.
For simplicity, this serves the full image. In production, you should
generate and cache thumbnails.
"""
logger.debug(f"Serving preview for attachment: {attachment_id}")
try:
file_path = attachment_store.get_file_path(attachment_id)
if not file_path.exists():
return JSONResponse(status_code=404, content={"error": "File not found"})
# Determine media type from file extension or attachment metadata
# For simplicity, we'll try to load from the store
try:
attachment = await data_store.load_attachment(attachment_id, {"user_id": DEFAULT_USER_ID})
media_type = attachment.mime_type
except Exception:
# Default to binary if we can't determine
media_type = "application/octet-stream"
return FileResponse(file_path, media_type=media_type)
except Exception as e:
logger.error(f"Error serving preview for attachment {attachment_id}: {e}", exc_info=True)
return JSONResponse(status_code=500, content={"error": str(e)})
if __name__ == "__main__":
# Run the server
logger.info(f"Starting ChatKit Weather Agent server on {SERVER_HOST}:{SERVER_PORT}")
uvicorn.run(app, host=SERVER_HOST, port=SERVER_PORT, log_level="info")
python/samples/demos/chatkit-integration/attachment_store.py
+121
View File
@@ -0,0 +1,121 @@
# Copyright (c) Microsoft. All rights reserved.
"""File-based AttachmentStore implementation for ChatKit.
This module provides a simple AttachmentStore implementation that stores
uploaded files on the local filesystem. In production, you should use
cloud storage like S3, Azure Blob Storage, or Google Cloud Storage.
"""
from pathlib import Path
from typing import Any, TYPE_CHECKING
from chatkit.store import AttachmentStore
from chatkit.types import Attachment, AttachmentCreateParams, FileAttachment, ImageAttachment
from pydantic import AnyUrl
if TYPE_CHECKING:
from store import SQLiteStore
class FileBasedAttachmentStore(AttachmentStore[dict[str, Any]]):
"""File-based AttachmentStore that stores files on local disk.
This implementation stores uploaded files in a local directory and provides
upload URLs that point to the FastAPI upload endpoint. It supports both
image and file attachments.
Features:
- Stores files in a local uploads directory
- Generates upload URLs for two-phase upload
- Generates preview URLs for images
- Proper cleanup on deletion
Note: This is for demonstration purposes. In production, use cloud storage
with signed URLs for better security and scalability.
"""
def __init__(
self,
uploads_dir: str = "./uploads",
base_url: str = "http://localhost:8001",
data_store: "SQLiteStore | None" = None,
):
"""Initialize the file-based attachment store.
Args:
uploads_dir: Directory where uploaded files will be stored
base_url: Base URL for generating upload and preview URLs
data_store: Optional data store to persist attachment metadata
"""
self.uploads_dir = Path(uploads_dir)
self.base_url = base_url.rstrip("/")
self.data_store = data_store
# Create uploads directory if it doesn't exist
self.uploads_dir.mkdir(parents=True, exist_ok=True)
def get_file_path(self, attachment_id: str) -> Path:
"""Get the filesystem path for an attachment."""
return self.uploads_dir / attachment_id
async def delete_attachment(self, attachment_id: str, context: dict[str, Any]) -> None:
"""Delete an attachment and its file from disk."""
file_path = self.get_file_path(attachment_id)
if file_path.exists():
file_path.unlink()
async def create_attachment(
self, input: AttachmentCreateParams, context: dict[str, Any]
) -> Attachment:
"""Create an attachment with upload URL for two-phase upload.
This creates the attachment metadata and returns upload URLs that
the client will use to POST the actual file bytes.
"""
# Generate unique ID for this attachment
attachment_id = self.generate_attachment_id(input.mime_type, context)
# Generate upload URL that points to our FastAPI upload endpoint
upload_url = f"{self.base_url}/upload/{attachment_id}"
# Create appropriate attachment type based on MIME type
if input.mime_type.startswith("image/"):
# For images, also provide a preview URL
preview_url = f"{self.base_url}/preview/{attachment_id}"
attachment = ImageAttachment(
id=attachment_id,
type="image",
mime_type=input.mime_type,
name=input.name,
upload_url=AnyUrl(upload_url),
preview_url=AnyUrl(preview_url),
)
else:
# For files, just provide upload URL
attachment = FileAttachment(
id=attachment_id,
type="file",
mime_type=input.mime_type,
name=input.name,
upload_url=AnyUrl(upload_url),
)
# Save attachment metadata to data store so it's available during upload
if self.data_store is not None:
await self.data_store.save_attachment(attachment, context)
return attachment
async def read_attachment_bytes(self, attachment_id: str) -> bytes:
"""Read the raw bytes of an uploaded attachment.
This is used by the ThreadItemConverter to create base64-encoded
content for sending to the Agent Framework.
"""
file_path = self.get_file_path(attachment_id)
if not file_path.exists():
raise FileNotFoundError(f"Attachment {attachment_id} not found on disk")
return file_path.read_bytes()
python/samples/demos/chatkit-integration/frontend/index.html
+52
View File
@@ -0,0 +1,52 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>ChatKit + Agent Framework Demo</title>
<script src="https://cdn.platform.openai.com/deployments/chatkit/chatkit.js"></script>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, 'Helvetica Neue', Arial, sans-serif;
height: 100vh;
display: flex;
flex-direction: column;
}
header {
padding: 1rem;
background: #f5f5f5;
border-bottom: 1px solid #ddd;
}
h1 {
font-size: 1.5rem;
margin-bottom: 0.5rem;
}
p {
color: #666;
font-size: 0.9rem;
}
#root {
flex: 1;
overflow: hidden;
}
</style>
</head>
<body>
<header>
<h1>ChatKit + Agent Framework Demo</h1>
<p>Simple weather assistant powered by Agent Framework and ChatKit</p>
</header>
<div id="root"></div>
<script type="module" src="/src/main.tsx"></script>
</body>
</html>
python/samples/demos/chatkit-integration/frontend/package-lock.json
Generated
+1437
View File
File diff suppressed because it is too large Load Diff
python/samples/demos/chatkit-integration/frontend/package.json
+27
View File
@@ -0,0 +1,27 @@
{
"name": "chatkit-agent-framework-demo",
"version": "0.1.0",
"private": true,
"type": "module",
"scripts": {
"dev": "vite",
"build": "vite build",
"preview": "vite preview"
},
"engines": {
"node": ">=18.18",
"npm": ">=9"
},
"dependencies": {
"@openai/chatkit-react": "^0",
"react": "^19.2.0",
"react-dom": "^19.2.0"
},
"devDependencies": {
"@types/react": "^19.2.0",
"@types/react-dom": "^19.2.0",
"@vitejs/plugin-react-swc": "^3.5.0",
"typescript": "^5.4.0",
"vite": "^7.1.9"
}
}
python/samples/demos/chatkit-integration/frontend/src/App.tsx
+33
View File
@@ -0,0 +1,33 @@
import { ChatKit, useChatKit } from "@openai/chatkit-react";
const CHATKIT_API_URL = "/chatkit";
const CHATKIT_API_DOMAIN_KEY =
import.meta.env.VITE_CHATKIT_API_DOMAIN_KEY ?? "domain_pk_localhost_dev";
export default function App() {
const chatkit = useChatKit({
api: {
url: CHATKIT_API_URL,
domainKey: CHATKIT_API_DOMAIN_KEY,
uploadStrategy: { type: "two_phase" },
},
startScreen: {
greeting: "Hello! I'm your weather and image analysis assistant. Ask me about the weather in any location or upload images for me to analyze.",
prompts: [
{ label: "Weather in New York", prompt: "What's the weather in New York?" },
{ label: "Select City to Get Weather", prompt: "Show me the city selector for weather" },
{ label: "Current Time", prompt: "What time is it?" },
{ label: "Analyze an Image", prompt: "I'll upload an image for you to analyze" },
],
},
composer: {
placeholder: "Ask about weather or upload an image...",
attachments: {
enabled: true,
accept: { "image/*": [".png", ".jpg", ".jpeg", ".gif", ".webp"] },
},
},
});
return <ChatKit control={chatkit.control} style={{ height: "100%" }} />;
}
python/samples/demos/chatkit-integration/frontend/src/main.tsx
+15
View File
@@ -0,0 +1,15 @@
import { StrictMode } from "react";
import { createRoot } from "react-dom/client";
import App from "./App";
const container = document.getElementById("root");
if (!container) {
throw new Error("Root element with id 'root' not found");
}
createRoot(container).render(
<StrictMode>
<App />
</StrictMode>,
);
python/samples/demos/chatkit-integration/frontend/src/vite-env.d.ts
Vendored
+1
View File
@@ -0,0 +1 @@
/// <reference types="vite/client" />
python/samples/demos/chatkit-integration/frontend/tsconfig.json
+21
View File
@@ -0,0 +1,21 @@
{
"compilerOptions": {
"target": "ES2020",
"useDefineForClassFields": true,
"lib": ["ES2020", "DOM", "DOM.Iterable"],
"module": "ESNext",
"skipLibCheck": true,
"moduleResolution": "bundler",
"allowImportingTsExtensions": true,
"resolveJsonModule": true,
"isolatedModules": true,
"noEmit": true,
"jsx": "react-jsx",
"strict": true,
"noUnusedLocals": true,
"noUnusedParameters": true,
"noFallthroughCasesInSwitch": true
},
"include": ["src"],
"references": [{ "path": "./tsconfig.node.json" }]
}
python/samples/demos/chatkit-integration/frontend/tsconfig.node.json
+10
View File
@@ -0,0 +1,10 @@
{
"compilerOptions": {
"composite": true,
"skipLibCheck": true,
"module": "ESNext",
"moduleResolution": "bundler",
"allowSyntheticDefaultImports": true
},
"include": ["vite.config.ts"]
}
python/samples/demos/chatkit-integration/frontend/vite.config.ts
+24
View File
@@ -0,0 +1,24 @@
import { defineConfig } from "vite";
import react from "@vitejs/plugin-react-swc";
const backendTarget = process.env.BACKEND_URL ?? "http://127.0.0.1:8001";
export default defineConfig({
plugins: [react()],
server: {
host: "0.0.0.0",
port: 5171,
proxy: {
"/chatkit": {
target: backendTarget,
changeOrigin: true,
},
},
// For production deployments, you need to add your public domains to this list
allowedHosts: [
// You can remove these examples added just to demonstrate how to configure the allowlist
".ngrok.io",
".trycloudflare.com",
],
},
});
python/samples/demos/chatkit-integration/store.py
+361
View File
@@ -0,0 +1,361 @@
# Copyright (c) Microsoft. All rights reserved.
"""SQLite-based store implementation for ChatKit data persistence.
This module provides a complete Store implementation using SQLite for data persistence.
It includes proper thread safety, user isolation, and follows the ChatKit Store protocol.
"""
import sqlite3
import uuid
from typing import Any
from chatkit.store import Store, NotFoundError
from chatkit.types import (
Attachment,
Page,
ThreadItem,
ThreadMetadata,
)
from pydantic import BaseModel
class ThreadData(BaseModel):
"""Model for serializing thread data to SQLite."""
thread: ThreadMetadata
class ItemData(BaseModel):
"""Model for serializing thread item data to SQLite."""
item: ThreadItem
class AttachmentData(BaseModel):
"""Model for serializing attachment data to SQLite."""
attachment: Attachment
class SQLiteStore(Store[dict[str, Any]]):
"""SQLite-based store implementation for ChatKit data.
This implementation follows the pattern from the ChatKit Python tests
and provides persistent storage for threads, messages, and attachments.
Features:
- Thread-safe SQLite connections with WAL mode
- User isolation for multi-tenant support
- Proper error handling and transaction management
- Complete Store protocol implementation
Note: This is for demonstration purposes. In production, you should
implement proper error handling, connection pooling, and migration strategies.
"""
def __init__(self, db_path: str | None = None):
self.db_path = db_path or "chatkit_demo.db" # Use file-based DB for demo
self._create_tables()
def _create_connection(self):
# Enable thread safety and WAL mode for better concurrent access
conn = sqlite3.connect(self.db_path, check_same_thread=False)
conn.execute("PRAGMA journal_mode=WAL")
return conn
def _create_tables(self):
with self._create_connection() as conn:
# Create threads table
conn.execute(
"""CREATE TABLE IF NOT EXISTS threads (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
created_at TEXT NOT NULL,
data TEXT NOT NULL
)"""
)
# Create items table
conn.execute(
"""CREATE TABLE IF NOT EXISTS items (
id TEXT PRIMARY KEY,
thread_id TEXT NOT NULL,
user_id TEXT NOT NULL,
created_at TEXT NOT NULL,
data TEXT NOT NULL
)"""
)
# Create attachments table
conn.execute(
"""CREATE TABLE IF NOT EXISTS attachments (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
data TEXT NOT NULL
)"""
)
conn.commit()
def generate_thread_id(self, context: dict[str, Any]) -> str:
return f"thr_{uuid.uuid4().hex[:8]}"
def generate_item_id(
self,
item_type: str,
thread: ThreadMetadata,
context: dict[str, Any],
) -> str:
prefix_map = {
"message": "msg",
"tool_call": "tc",
"task": "tsk",
"workflow": "wf",
"attachment": "atc",
}
prefix = prefix_map.get(item_type, "itm")
return f"{prefix}_{uuid.uuid4().hex[:8]}"
async def load_thread(self, thread_id: str, context: dict[str, Any]) -> ThreadMetadata:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
cursor = conn.execute(
"SELECT data FROM threads WHERE id = ? AND user_id = ?",
(thread_id, user_id),
).fetchone()
if cursor is None:
raise NotFoundError(f"Thread {thread_id} not found")
thread_data = ThreadData.model_validate_json(cursor[0])
return thread_data.thread
async def save_thread(self, thread: ThreadMetadata, context: dict[str, Any]) -> None:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
thread_data = ThreadData(thread=thread)
# Replace existing thread data
conn.execute(
"DELETE FROM threads WHERE id = ? AND user_id = ?",
(thread.id, user_id),
)
conn.execute(
"INSERT INTO threads (id, user_id, created_at, data) VALUES (?, ?, ?, ?)",
(
thread.id,
user_id,
thread.created_at.isoformat(),
thread_data.model_dump_json(),
),
)
conn.commit()
async def load_thread_items(
self,
thread_id: str,
after: str | None,
limit: int,
order: str,
context: dict[str, Any],
) -> Page[ThreadItem]:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
created_after: str | None = None
if after:
after_cursor = conn.execute(
"SELECT created_at FROM items WHERE id = ? AND user_id = ?",
(after, user_id),
).fetchone()
if after_cursor is None:
raise NotFoundError(f"Item {after} not found")
created_after = after_cursor[0]
query = """
SELECT data FROM items
WHERE thread_id = ? AND user_id = ?
"""
params: list[Any] = [thread_id, user_id]
if created_after:
query += " AND created_at > ?" if order == "asc" else " AND created_at < ?"
params.append(created_after)
query += f" ORDER BY created_at {order} LIMIT ?"
params.append(limit + 1)
items_cursor = conn.execute(query, params).fetchall()
items = [
ItemData.model_validate_json(row[0]).item for row in items_cursor
]
has_more = len(items) > limit
if has_more:
items = items[:limit]
return Page[ThreadItem](
data=items,
has_more=has_more,
after=items[-1].id if items else None
)
async def save_attachment(self, attachment: Attachment, context: dict[str, Any]) -> None:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
attachment_data = AttachmentData(attachment=attachment)
conn.execute(
"INSERT OR REPLACE INTO attachments (id, user_id, data) VALUES (?, ?, ?)",
(
attachment.id,
user_id,
attachment_data.model_dump_json(),
),
)
conn.commit()
async def load_attachment(self, attachment_id: str, context: dict[str, Any]) -> Attachment:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
cursor = conn.execute(
"SELECT data FROM attachments WHERE id = ? AND user_id = ?",
(attachment_id, user_id),
).fetchone()
if cursor is None:
raise NotFoundError(f"Attachment {attachment_id} not found")
attachment_data = AttachmentData.model_validate_json(cursor[0])
return attachment_data.attachment
async def delete_attachment(self, attachment_id: str, context: dict[str, Any]) -> None:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
conn.execute(
"DELETE FROM attachments WHERE id = ? AND user_id = ?",
(attachment_id, user_id),
)
conn.commit()
async def load_threads(
self,
limit: int,
after: str | None,
order: str,
context: dict[str, Any],
) -> Page[ThreadMetadata]:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
created_after: str | None = None
if after:
after_cursor = conn.execute(
"SELECT created_at FROM threads WHERE id = ? AND user_id = ?",
(after, user_id),
).fetchone()
if after_cursor is None:
raise NotFoundError(f"Thread {after} not found")
created_after = after_cursor[0]
query = "SELECT data FROM threads WHERE user_id = ?"
params: list[Any] = [user_id]
if created_after:
query += " AND created_at > ?" if order == "asc" else " AND created_at < ?"
params.append(created_after)
query += f" ORDER BY created_at {order} LIMIT ?"
params.append(limit + 1)
threads_cursor = conn.execute(query, params).fetchall()
threads = [
ThreadData.model_validate_json(row[0]).thread for row in threads_cursor
]
has_more = len(threads) > limit
if has_more:
threads = threads[:limit]
return Page[ThreadMetadata](
data=threads,
has_more=has_more,
after=threads[-1].id if threads else None
)
async def add_thread_item(
self, thread_id: str, item: ThreadItem, context: dict[str, Any]
) -> None:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
item_data = ItemData(item=item)
conn.execute(
"INSERT INTO items (id, thread_id, user_id, created_at, data) VALUES (?, ?, ?, ?, ?)",
(
item.id,
thread_id,
user_id,
item.created_at.isoformat(),
item_data.model_dump_json(),
),
)
conn.commit()
async def save_item(self, thread_id: str, item: ThreadItem, context: dict[str, Any]) -> None:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
item_data = ItemData(item=item)
conn.execute(
"UPDATE items SET data = ? WHERE id = ? AND thread_id = ? AND user_id = ?",
(
item_data.model_dump_json(),
item.id,
thread_id,
user_id,
),
)
conn.commit()
async def load_item(self, thread_id: str, item_id: str, context: dict[str, Any]) -> ThreadItem:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
cursor = conn.execute(
"SELECT data FROM items WHERE id = ? AND thread_id = ? AND user_id = ?",
(item_id, thread_id, user_id),
).fetchone()
if cursor is None:
raise NotFoundError(f"Item {item_id} not found in thread {thread_id}")
item_data = ItemData.model_validate_json(cursor[0])
return item_data.item
async def delete_thread(self, thread_id: str, context: dict[str, Any]) -> None:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
conn.execute(
"DELETE FROM threads WHERE id = ? AND user_id = ?",
(thread_id, user_id),
)
conn.execute(
"DELETE FROM items WHERE thread_id = ? AND user_id = ?",
(thread_id, user_id),
)
conn.commit()
async def delete_thread_item(
self, thread_id: str, item_id: str, context: dict[str, Any]
) -> None:
user_id = context.get("user_id", "demo_user")
with self._create_connection() as conn:
conn.execute(
"DELETE FROM items WHERE id = ? AND thread_id = ? AND user_id = ?",
(item_id, thread_id, user_id),
)
conn.commit()
python/samples/demos/chatkit-integration/weather_widget.py
+437
View File
@@ -0,0 +1,437 @@
# Copyright (c) Microsoft. All rights reserved.
"""Weather widget rendering for ChatKit integration sample."""
import base64
from dataclasses import dataclass
from chatkit.actions import ActionConfig
from chatkit.widgets import Box, Button, Card, Col, Image, Row, Text, Title, WidgetRoot
WEATHER_ICON_COLOR = "#1D4ED8"
WEATHER_ICON_ACCENT = "#DBEAFE"
# Popular cities for the selector
POPULAR_CITIES = [
{"value": "seattle", "label": "Seattle, WA", "description": "Pacific Northwest"},
{"value": "new_york", "label": "New York, NY", "description": "East Coast"},
{"value": "san_francisco", "label": "San Francisco, CA", "description": "Bay Area"},
{"value": "chicago", "label": "Chicago, IL", "description": "Midwest"},
{"value": "miami", "label": "Miami, FL", "description": "Southeast"},
{"value": "austin", "label": "Austin, TX", "description": "Southwest"},
{"value": "boston", "label": "Boston, MA", "description": "New England"},
{"value": "denver", "label": "Denver, CO", "description": "Mountain West"},
{"value": "portland", "label": "Portland, OR", "description": "Pacific Northwest"},
{"value": "atlanta", "label": "Atlanta, GA", "description": "Southeast"},
]
# Mapping from city values to display names for weather queries
CITY_VALUE_TO_NAME = {city["value"]: city["label"] for city in POPULAR_CITIES}
def _sun_svg() -> str:
"""Generate SVG for sunny weather icon."""
color = WEATHER_ICON_COLOR
accent = WEATHER_ICON_ACCENT
return (
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none">'
f'<circle cx="32" cy="32" r="13" fill="{accent}" stroke="{color}" stroke-width="3"/>'
f'<g stroke="{color}" stroke-width="3" stroke-linecap="round">'
'<line x1="32" y1="8" x2="32" y2="16"/>'
'<line x1="32" y1="48" x2="32" y2="56"/>'
'<line x1="8" y1="32" x2="16" y2="32"/>'
'<line x1="48" y1="32" x2="56" y2="32"/>'
'<line x1="14.93" y1="14.93" x2="20.55" y2="20.55"/>'
'<line x1="43.45" y1="43.45" x2="49.07" y2="49.07"/>'
'<line x1="14.93" y1="49.07" x2="20.55" y2="43.45"/>'
'<line x1="43.45" y1="20.55" x2="49.07" y2="14.93"/>'
"</g>"
"</svg>"
)
def _cloud_svg() -> str:
"""Generate SVG for cloudy weather icon."""
color = WEATHER_ICON_COLOR
accent = WEATHER_ICON_ACCENT
return (
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none">'
f'<path d="M22 46H44C50.075 46 55 41.075 55 35S50.075 24 44 24H42.7C41.2 16.2 34.7 10 26.5 10 18 10 11.6 16.1 11 24.3 6.5 25.6 3 29.8 3 35s4.925 11 11 11h8Z" '
f'fill="{accent}" stroke="{color}" stroke-width="3" stroke-linejoin="round"/>'
"</svg>"
)
def _rain_svg() -> str:
"""Generate SVG for rainy weather icon."""
color = WEATHER_ICON_COLOR
accent = WEATHER_ICON_ACCENT
return (
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none">'
f'<path d="M22 40H44C50.075 40 55 35.075 55 29S50.075 18 44 18H42.7C41.2 10.2 34.7 4 26.5 4 18 4 11.6 10.1 11 18.3 6.5 19.6 3 23.8 3 29s4.925 11 11 11h8Z" '
f'fill="{accent}" stroke="{color}" stroke-width="3" stroke-linejoin="round"/>'
f'<g stroke="{color}" stroke-width="3" stroke-linecap="round">'
'<line x1="20" y1="48" x2="24" y2="56"/>'
'<line x1="30" y1="50" x2="34" y2="58"/>'
'<line x1="40" y1="48" x2="44" y2="56"/>'
"</g>"
"</svg>"
)
def _storm_svg() -> str:
"""Generate SVG for stormy weather icon."""
color = WEATHER_ICON_COLOR
accent = WEATHER_ICON_ACCENT
return (
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none">'
f'<path d="M22 40H44C50.075 40 55 35.075 55 29S50.075 18 44 18H42.7C41.2 10.2 34.7 4 26.5 4 18 4 11.6 10.1 11 18.3 6.5 19.6 3 23.8 3 29s4.925 11 11 11h8Z" '
f'fill="{accent}" stroke="{color}" stroke-width="3" stroke-linejoin="round"/>'
f'<path d="M34 46L28 56H34L30 64L42 50H36L40 46Z" '
f'fill="{color}" stroke="{color}" stroke-width="2" stroke-linejoin="round"/>'
"</svg>"
)
def _snow_svg() -> str:
"""Generate SVG for snowy weather icon."""
color = WEATHER_ICON_COLOR
accent = WEATHER_ICON_ACCENT
return (
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none">'
f'<path d="M22 40H44C50.075 40 55 35.075 55 29S50.075 18 44 18H42.7C41.2 10.2 34.7 4 26.5 4 18 4 11.6 10.1 11 18.3 6.5 19.6 3 23.8 3 29s4.925 11 11 11h8Z" '
f'fill="{accent}" stroke="{color}" stroke-width="3" stroke-linejoin="round"/>'
f'<g stroke="{color}" stroke-width="2" stroke-linecap="round">'
'<line x1="20" y1="48" x2="20" y2="56"/>'
'<line x1="17" y1="51" x2="23" y2="53"/>'
'<line x1="17" y1="53" x2="23" y2="51"/>'
'<line x1="36" y1="48" x2="36" y2="56"/>'
'<line x1="33" y1="51" x2="39" y2="53"/>'
'<line x1="33" y1="53" x2="39" y2="51"/>'
"</g>"
"</svg>"
)
def _fog_svg() -> str:
"""Generate SVG for foggy weather icon."""
color = WEATHER_ICON_COLOR
accent = WEATHER_ICON_ACCENT
return (
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none">'
f'<path d="M22 40H44C50.075 40 55 35.075 55 29S50.075 18 44 18H42.7C41.2 10.2 34.7 4 26.5 4 18 4 11.6 10.1 11 18.3 6.5 19.6 3 23.8 3 29s4.925 11 11 11h8Z" '
f'fill="{accent}" stroke="{color}" stroke-width="3" stroke-linejoin="round"/>'
f'<g stroke="{color}" stroke-width="3" stroke-linecap="round">'
'<line x1="18" y1="50" x2="42" y2="50"/>'
'<line x1="24" y1="56" x2="48" y2="56"/>'
"</g>"
"</svg>"
)
def _encode_svg(svg: str) -> str:
"""Encode SVG as base64 data URI."""
encoded = base64.b64encode(svg.encode("utf-8")).decode("ascii")
return f"data:image/svg+xml;base64,{encoded}"
# Weather condition to icon mapping
WEATHER_ICONS = {
"sunny": _encode_svg(_sun_svg()),
"cloudy": _encode_svg(_cloud_svg()),
"rainy": _encode_svg(_rain_svg()),
"stormy": _encode_svg(_storm_svg()),
"snowy": _encode_svg(_snow_svg()),
"foggy": _encode_svg(_fog_svg()),
}
DEFAULT_WEATHER_ICON = _encode_svg(_cloud_svg())
@dataclass
class WeatherData:
"""Weather data container."""
location: str
condition: str
temperature: int
humidity: int
wind_speed: int
def render_weather_widget(data: WeatherData) -> WidgetRoot:
"""Render a weather widget from weather data.
Args:
data: WeatherData containing weather information
Returns:
A ChatKit WidgetRoot (Card) displaying the weather information
"""
# Get weather icon
weather_icon_src = WEATHER_ICONS.get(data.condition.lower(), DEFAULT_WEATHER_ICON)
# Build the widget
header = Box(
padding=5,
background="surface-tertiary",
children=[
Row(
justify="between",
align="center",
children=[
Col(
align="start",
gap=1,
children=[
Text(
value=data.location,
size="lg",
weight="semibold",
),
Text(
value="Current conditions",
color="tertiary",
size="xs",
),
],
),
Box(
padding=3,
radius="full",
background="blue-100",
children=[
Image(
src=weather_icon_src,
alt=data.condition,
size=28,
fit="contain",
)
],
),
],
),
Row(
align="start",
gap=4,
children=[
Title(
value=f"{data.temperature}°C",
size="lg",
weight="semibold",
),
Col(
align="start",
gap=1,
children=[
Text(
value=data.condition.title(),
color="secondary",
size="sm",
weight="medium",
),
],
),
],
),
],
)
# Details section
details = Box(
padding=5,
gap=4,
children=[
Text(value="Weather details", weight="semibold", size="sm"),
Row(
gap=3,
wrap="wrap",
children=[
_detail_chip("Humidity", f"{data.humidity}%"),
_detail_chip("Wind", f"{data.wind_speed} km/h"),
],
),
],
)
return Card(
key="weather",
padding=0,
children=[header, details],
)
def _detail_chip(label: str, value: str) -> Box:
"""Create a detail chip widget component."""
return Box(
padding=3,
radius="xl",
background="surface-tertiary",
width=150,
minWidth=150,
maxWidth=150,
minHeight=80,
maxHeight=80,
flex="0 0 auto",
children=[
Col(
align="stretch",
gap=2,
children=[
Text(value=label, size="xs", weight="medium", color="tertiary"),
Row(
justify="center",
margin={"top": 2},
children=[Text(value=value, weight="semibold", size="lg")],
),
],
)
],
)
def weather_widget_copy_text(data: WeatherData) -> str:
"""Generate plain text representation of weather data.
Args:
data: WeatherData containing weather information
Returns:
Plain text description for copy/paste functionality
"""
return (
f"Weather in {data.location}:\n"
f"• Condition: {data.condition.title()}\n"
f"• Temperature: {data.temperature}°C\n"
f"• Humidity: {data.humidity}%\n"
f"• Wind: {data.wind_speed} km/h"
)
def render_city_selector_widget() -> WidgetRoot:
"""Render an interactive city selector widget.
This widget displays popular cities as a visual selection interface.
Users can click or ask about any city to get weather information.
Returns:
A ChatKit WidgetRoot (Card) with city selection display
"""
# Create location icon SVG
location_icon = _encode_svg(
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" fill="none">'
f'<path d="M32 8c-8.837 0-16 7.163-16 16 0 12 16 32 16 32s16-20 16-32c0-8.837-7.163-16-16-16z" '
f'fill="{WEATHER_ICON_ACCENT}" stroke="{WEATHER_ICON_COLOR}" stroke-width="3" stroke-linejoin="round"/>'
f'<circle cx="32" cy="24" r="6" fill="{WEATHER_ICON_COLOR}"/>'
"</svg>"
)
# Header section
header = Box(
padding=5,
background="surface-tertiary",
children=[
Row(
gap=3,
align="center",
children=[
Box(
padding=3,
radius="full",
background="blue-100",
children=[
Image(
src=location_icon,
alt="Location",
size=28,
fit="contain",
)
],
),
Col(
align="start",
gap=1,
children=[
Title(
value="Popular Cities",
size="md",
weight="semibold",
),
Text(
value="Select a city or ask about any location",
color="tertiary",
size="xs",
),
],
),
],
),
],
)
# Create city chips in a grid layout
city_chips: list[Button] = []
for city in POPULAR_CITIES:
# Create a button that sends an action to query weather for the selected city
chip = Button(
label=city["label"],
variant="outline",
size="md",
onClickAction=ActionConfig(
type="city_selected",
payload={"city_value": city["value"], "city_label": city["label"]},
handler="server", # Handle on server-side
),
)
city_chips.append(chip)
# Arrange in rows of 3
city_rows: list[Row] = []
for i in range(0, len(city_chips), 3):
row_chips: list[Button] = city_chips[i : i + 3]
city_rows.append(
Row(
gap=3,
wrap="wrap",
justify="start",
children=list(row_chips), # Convert to generic list
)
)
# Cities display section
cities_section = Box(
padding=5,
gap=3,
children=[
*city_rows,
Box(
padding=3,
radius="md",
background="blue-50",
children=[
Text(
value="💡 Click any city to get its weather, or ask about any other location!",
size="xs",
color="secondary",
),
],
),
],
)
return Card(
key="city_selector",
padding=0,
children=[header, cities_section],
)
def city_selector_copy_text() -> str:
"""Generate plain text representation of city selector.
Returns:
Plain text description for copy/paste functionality
"""
cities_list = "\n".join([f"• {city['label']}" for city in POPULAR_CITIES])
return f"Popular cities (click to get weather):\n{cities_list}\n\nYou can also ask about weather in any other location!"
python/uv.lock
Generated
+75
View File
@@ -33,6 +33,7 @@ members = [
"agent-framework-a2a",
"agent-framework-anthropic",
"agent-framework-azure-ai",
"agent-framework-chatkit",
"agent-framework-copilotstudio",
"agent-framework-core",
"agent-framework-devui",
@@ -79,6 +80,7 @@ dependencies = [
{ name = "agent-framework-a2a", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "agent-framework-anthropic", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "agent-framework-azure-ai", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "agent-framework-chatkit", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "agent-framework-copilotstudio", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "agent-framework-devui", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
@@ -121,6 +123,7 @@ requires-dist = [
{ name = "agent-framework-a2a", editable = "packages/a2a" },
{ name = "agent-framework-anthropic", editable = "packages/anthropic" },
{ name = "agent-framework-azure-ai", editable = "packages/azure-ai" },
{ name = "agent-framework-chatkit", editable = "packages/chatkit" },
{ name = "agent-framework-copilotstudio", editable = "packages/copilotstudio" },
{ name = "agent-framework-core", editable = "packages/core" },
{ name = "agent-framework-devui", editable = "packages/devui" },
@@ -205,6 +208,21 @@ requires-dist = [
{ name = "azure-ai-projects", specifier = ">=1.0.0b11" },
]
[[package]]
name = "agent-framework-chatkit"
version = "1.0.0b251001"
source = { editable = "packages/chatkit" }
dependencies = [
{ name = "agent-framework-core", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "openai-chatkit", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
]
[package.metadata]
requires-dist = [
{ name = "agent-framework-core", editable = "packages/core" },
{ name = "openai-chatkit", specifier = ">=1.1.0,<2.0.0" },
]
[[package]]
name = "agent-framework-copilotstudio"
version = "1.0.0b251104"
@@ -2086,6 +2104,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e3/a5/6ddab2b4c112be95601c13428db1d8b6608a8b6039816f2ba09c346c08fc/greenlet-3.2.4-cp314-cp314-win_amd64.whl", hash = "sha256:e37ab26028f12dbb0ff65f29a8d3d44a765c61e729647bf2ddfbbed621726f01", size = 303425, upload-time = "2025-08-07T13:32:27.59Z" },
]
[[package]]
name = "griffe"
version = "1.14.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/ec/d7/6c09dd7ce4c7837e4cdb11dce980cb45ae3cd87677298dc3b781b6bce7d3/griffe-1.14.0.tar.gz", hash = "sha256:9d2a15c1eca966d68e00517de5d69dd1bc5c9f2335ef6c1775362ba5b8651a13", size = 424684, upload-time = "2025-09-05T15:02:29.167Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/b1/9ff6578d789a89812ff21e4e0f80ffae20a65d5dd84e7a17873fe3b365be/griffe-1.14.0-py3-none-any.whl", hash = "sha256:0e9d52832cccf0f7188cfe585ba962d2674b241c01916d780925df34873bceb0", size = 144439, upload-time = "2025-09-05T15:02:27.511Z" },
]
[[package]]
name = "grpcio"
version = "1.76.0"
@@ -3497,6 +3527,39 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/1d/2a/7dd3d207ec669cacc1f186fd856a0f61dbc255d24f6fdc1a6715d6051b0f/openai-1.109.1-py3-none-any.whl", hash = "sha256:6bcaf57086cf59159b8e27447e4e7dd019db5d29a438072fbd49c290c7e65315", size = 948627, upload-time = "2025-09-24T13:00:50.754Z" },
]
[[package]]
name = "openai-agents"
version = "0.3.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "griffe", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "mcp", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "openai", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "pydantic", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "requests", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "types-requests", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "typing-extensions", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a4/37/2b4f828840d3ff32d82b813c3371ec9ee26b3b8dc6b4acbb7a4a579f617a/openai_agents-0.3.3.tar.gz", hash = "sha256:b016381a6890e1cb6879eb23c53c35f8c2312be1117f1cd4e4b5e2463150839f", size = 1816230, upload-time = "2025-09-30T23:20:24.22Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/65/59/fd49fd2c3184c0d5fedb8c9c456ae9852154828bca7ee69dce004ea83188/openai_agents-0.3.3-py3-none-any.whl", hash = "sha256:aa2c74e010b923c09f166e63a51fae8c850c62df8581b84bafcbe5bd208d1505", size = 210893, upload-time = "2025-09-30T23:20:22.037Z" },
]
[[package]]
name = "openai-chatkit"
version = "1.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "openai", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "openai-agents", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "pydantic", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
{ name = "uvicorn", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/4a/19/9948f2996c224aff01f6ef415784042c3d710c1e950937b16d9a2c07e47e/openai_chatkit-1.1.0.tar.gz", hash = "sha256:5594341aab29b56fd3396e8d3ad1962ebdb8c44f062a8e315663ac8cf1371c6b", size = 49480, upload-time = "2025-11-03T22:50:05.089Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/71/82/07db74ee63d54f3cadab3baaa1534bef0d3699a94d2618c76050cccb0cfe/openai_chatkit-1.1.0-py3-none-any.whl", hash = "sha256:e78f021899fbef1323f3adc3a686f9fe5ee184cd997799a917e9013833e760ba", size = 35424, upload-time = "2025-11-03T22:50:03.788Z" },
]
[[package]]
name = "opentelemetry-api"
version = "1.38.0"
@@ -5919,6 +5982,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/5e/dd/5cbf31f402f1cc0ab087c94d4669cfa55bd1e818688b910631e131d74e75/typer_slim-0.20.0-py3-none-any.whl", hash = "sha256:f42a9b7571a12b97dddf364745d29f12221865acef7a2680065f9bb29c7dc89d", size = 47087, upload-time = "2025-10-20T17:03:44.546Z" },
]
[[package]]
name = "types-requests"
version = "2.32.4.20250913"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "urllib3", marker = "sys_platform == 'darwin' or sys_platform == 'linux' or sys_platform == 'win32'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/36/27/489922f4505975b11de2b5ad07b4fe1dca0bca9be81a703f26c5f3acfce5/types_requests-2.32.4.20250913.tar.gz", hash = "sha256:abd6d4f9ce3a9383f269775a9835a4c24e5cd6b9f647d64f88aa4613c33def5d", size = 23113, upload-time = "2025-09-13T02:40:02.309Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/20/9a227ea57c1285986c4cf78400d0a91615d25b24e257fd9e2969606bdfae/types_requests-2.32.4.20250913-py3-none-any.whl", hash = "sha256:78c9c1fffebbe0fa487a418e0fa5252017e9c60d1a2da394077f1780f655d7e1", size = 20658, upload-time = "2025-09-13T02:40:01.115Z" },
]
[[package]]
name = "typing-extensions"
version = "4.15.0"