Skip to content

Python: Fix ChatKit integration sample startup and weather widgets #9579

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Naveen Chatlapalli (1aifanatic) wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Python: Fix ChatKit integration sample startup and weather widgets #9579

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 43 additions & 66 deletions python/samples/05-end-to-end/chatkit-integration/app.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
import uvicorn

# Agent Framework imports
from agent_framework import Agent, AgentResponseUpdate, FunctionResultContent, Message, Role, tool
from agent_framework import Agent, AgentResponseUpdate, FunctionInvocationContext, Message, tool
from agent_framework.foundry import FoundryChatClient

# Agent Framework ChatKit integration
Expand Down Expand Up @@ -105,15 +105,6 @@
logger = logging.getLogger(__name__)


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,
Expand Down Expand Up @@ -151,10 +142,13 @@ async def stream_widget(
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
ctx: FunctionInvocationContext,
) -> str:
"""Get the weather for a given location.

Returns a string description with embedded WeatherData for widget creation.
Returns a string description for the model. The structured WeatherData is appended to the
request-scoped ``weather_results`` list passed via ``function_invocation_kwargs`` so the
server can render a weather widget; it is never exposed to the model.
"""
logger.info(f"Fetching weather for location: {location}")

Expand All @@ -176,15 +170,19 @@ def get_weather(

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 = (
# Hand the structured data to the server for widget rendering
weather_results = ctx.kwargs.get("weather_results")
if isinstance(weather_results, list):
weather_results.append(weather_data)

# Return a plain-text description for the LLM
return (
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)


@tool(approval_mode="never_require")
Expand Down Expand Up @@ -297,7 +295,7 @@ async def _update_thread_title(

title_prompt = [
Message(
role=Role.USER,
role="user",
contents=[
(
f"Generate a very short, concise title (max 40 characters) for a conversation "
Expand Down Expand Up @@ -350,17 +348,15 @@ async def respond(
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
# 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
Expand All @@ -383,29 +379,22 @@ async def respond(

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)
# 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 extracts function results while passing through updates
# Create an intercepting stream that detects the city selector marker while passing through updates
async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
nonlocal weather_data, show_city_selector
nonlocal 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")
# 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
Expand All @@ -416,7 +405,8 @@ async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
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):
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)
Expand Down Expand Up @@ -472,44 +462,31 @@ async def action(
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
# Collect weather data from get_weather for this request
weather_results: list[WeatherData] = []

# Create an agent message asking about the weather
agent_messages = [Message(role=Role.USER, contents=[f"What's the weather in {city_label}?"])]
agent_messages = [Message(role="user", contents=[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(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
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
# 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},
)

# Stream updates as ChatKit events with interception
# Stream updates as ChatKit events
async for event in stream_agent_response(
intercept_stream(),
agent_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):
if weather_results:
weather_data = weather_results[-1]
logger.info(f"Creating weather widget for: {weather_data.location}")
# Create weather widget
widget = render_weather_widget(weather_data)
Expand Down
Loading