ChatKit

agent-framework-chatkit converte itens do thread OpenAI ChatKit em mensagens do Agent Framework e converte as atualizações transmitidas de agentes de volta em eventos ChatKit. Usa-o quando quiseres um frontend ChatKit com um backend Python do Agent Framework.

A integração proporciona:

• ThreadItemConverter para converter elementos e anexos da conversa do ChatKit.
• stream_agent_response() para converter atualizações de agentes transmitidas em eventos do ChatKit.
• simple_to_agent_input() para o caminho padrão de conversão de mensagens.

Pré-requisitos

• Python 3.11 ou posterior.
• Um framework web backend como o FastAPI.
• Node.js para o frontend do ChatKit.
• Um projeto Microsoft Foundry com um modelo implementado.
• Autenticação CLI do Azure (az login).
• Uma chave de domínio ChatKit para um domínio frontend de produção.

Instale o pacote

pip install agent-framework-chatkit agent-framework-foundry --pre

Configurar o modelo Foundry

A amostra cria o agente com FoundryChatClient e AzureCliCredential. Defina o endpoint do projeto e a implantação do modelo antes de iniciar o backend:

export FOUNDRY_PROJECT_ENDPOINT="https://<account>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL="<model-deployment>"

Criar um servidor ChatKit

Subclasse ChatKitServer, crie o agente Agent Framework e configure um conversor para itens de thread e anexos.

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,
    )

    logger.info("WeatherChatKitServer initialized")

Converter e transmitir respostas em fluxo

Carrega o histórico da thread, converte-o para mensagens do Agent Framework, executa o agente em modo streaming e gera eventos do ChatKit.

    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.
    """
    if input_user_message is None:
        logger.debug("Received None user message, skipping")
        return

    logger.info(f"Processing message for thread: {thread.id}")

    try:
        # Collect weather data from get_weather and track the city selector flag for this request
        weather_results: list[WeatherData] = []
        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. get_weather appends to weather_results.
        agent_stream = self.weather_agent.run(
            agent_messages,
            stream=True,
            function_invocation_kwargs={"weather_results": weather_results},
        )

        # Create an intercepting stream that detects the city selector marker while passing through updates
        async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
            nonlocal show_city_selector
            async for update in agent_stream:
                # Check for the city selector marker in function results
                for content in update.contents:
                    if content.type == "function_result" and content.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_results:
            weather_data = weather_results[-1]
            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):

Chame agent.run(..., stream=True) primeiro, depois passe esse fluxo de resposta e o ID do thread do ChatKit para stream_agent_response(...). Não passe o próprio agente para o conversor.

ThreadItemConverter preserva StructuredInputItem o estado e representa cada pergunta como respondida, omitida ou não respondida numa mensagem de utilizador. Redefina o método assíncrono structured_input_to_input() para personalizar a formatação ou ocultar respostas. Devolve uma Message, uma lista de mensagens ou None para omitir o item.

O conversor também preserva os valores concluídos GeneratedImageItem como contexto do modelo em curvas posteriores. A conversão padrão cria um utilizador Message com um prefácio de texto curto e o URI da imagem gerada. Os URIs de dados mantêm o seu tipo de multimédia incorporado, os URLs externos utilizam image/*, e os itens incompletos sem imagem são ignorados.

Redefina o método assíncrono generated_image_to_input() para personalizar o contexto da imagem gerada, por exemplo, para resolver o URL de uma imagem privada. Devolve uma Message, uma lista de mensagens ou None para omitir o item.

O exemplo completo também demonstra threads suportados por SQLite, carregamentos de ficheiros, armazenamento de anexos, ações e widgets interativos.

Warning

O frontend do ChatKit é carregado da CDN da OpenAI e envia pedidos para domínios da OpenAI. Neste momento, não pode ser auto-hospedado e não é adequado para ambientes isolados.

Passos seguintes

DevUI