agent-framework-chatkit 將 OpenAI ChatKit 執行緒項目轉換為代理框架訊息，並將串流代理更新轉回 ChatKit 事件。 當你想要 ChatKit 前端搭配 Agent Framework Python 後端時，可以用它。
整合提供：
-
ThreadItemConverter用於轉換 ChatKit 的執行緒項目與附件。 -
stream_agent_response()用於將串流客服更新轉換為 ChatKit 事件。 -
simple_to_agent_input()用於預設訊息轉換路徑。
先決條件
- Python 3.10 或更新版本。
- 後端網頁框架如 FastAPI。
- Node.js ChatKit 前端。
- 一個用於生產前端網域的 ChatKit 網域金鑰。
安裝套件
pip install agent-framework-chatkit --pre
建立 ChatKit 伺服器
子類 ChatKitServer，建立 Agent Framework 代理，並設定執行緒項目與附件的轉換器。
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 = Agent(
client=FoundryChatClient(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,
)
轉換與串流回應
載入執行緒歷史，轉換成 Agent Framework 訊息，將代理程式切換成串流模式，並產生 ChatKit 事件。
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
# Load full thread history from the store
thread_items_page = await self.store.load_thread_items(
thread_id=thread.id,
after=None,
limit=1000,
order="asc",
context=context,
)
thread_items = thread_items_page.data
# Convert ALL thread items to Agent Framework ChatMessages using ThreadItemConverter
# This ensures the agent has the full conversation context
agent_messages = await self.converter.to_agent_input(thread_items)
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(agent_messages, stream=True)
# Create an intercepting stream that extracts function results while passing through updates
async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
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
完整範例還展示了 SQLite 支援的執行緒、檔案上傳、附件儲存、動作以及互動式小工具。
Warning
ChatKit 前端是從 OpenAI 的 CDN 載入，並向 OpenAI 網域發出外撥請求。 目前無法自行架設，也不適合空隙環境。