From 39566d1dbf0cf76787884a382a1b48f95505c881 Mon Sep 17 00:00:00 2001 From: Copilot <198982749+Copilot@users.noreply.github.com> Date: Sun, 21 Sep 2025 21:42:33 -0700 Subject: [PATCH] Python: Add custom agent and chat client implementation examples (#849) * Initial plan * Add custom agent and chat client examples with complete implementations and documentation Co-authored-by: dmytrostruk <13853051+dmytrostruk@users.noreply.github.com> * Simplify custom examples per feedback: remove __init__.py, keep only EchoAgent/EchoingChatClient, add proper documentation, update README table format Co-authored-by: dmytrostruk <13853051+dmytrostruk@users.noreply.github.com> * Small fixes and formatting --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: dmytrostruk <13853051+dmytrostruk@users.noreply.github.com> --- .../getting_started/agents/custom/README.md | 26 +++ .../agents/custom/custom_agent.py | 215 ++++++++++++++++++ .../agents/custom/custom_chat_client.py | 183 +++++++++++++++ 3 files changed, 424 insertions(+) create mode 100644 python/samples/getting_started/agents/custom/README.md create mode 100644 python/samples/getting_started/agents/custom/custom_agent.py create mode 100644 python/samples/getting_started/agents/custom/custom_chat_client.py diff --git a/python/samples/getting_started/agents/custom/README.md b/python/samples/getting_started/agents/custom/README.md new file mode 100644 index 0000000000..62e426b7af --- /dev/null +++ b/python/samples/getting_started/agents/custom/README.md @@ -0,0 +1,26 @@ +# Custom Agent and Chat Client Examples + +This folder contains examples demonstrating how to implement custom agents and chat clients using the Microsoft Agent Framework. + +## Examples + +| File | Description | +|------|-------------| +| [`custom_agent.py`](custom_agent.py) | Shows how to create custom agents by extending the `BaseAgent` class. Demonstrates the `EchoAgent` implementation with both streaming and non-streaming responses, proper thread management, and message history handling. | +| [`custom_chat_client.py`](custom_chat_client.py) | Demonstrates how to create custom chat clients by extending the `BaseChatClient` class. Shows the `EchoingChatClient` implementation and how to integrate it with `ChatAgent` using the `create_agent()` method. | + +## Key Takeaways + +### Custom Agents +- Custom agents give you complete control over the agent's behavior +- You must implement both `run()` (for complete responses) and `run_stream()` (for streaming responses) +- Use `self._normalize_messages()` to handle different input message formats +- Use `self._notify_thread_of_new_messages()` to properly manage conversation history + +### Custom Chat Clients +- Custom chat clients allow you to integrate any backend service or create new LLM providers +- You must implement both `_inner_get_response()` and `_inner_get_streaming_response()` +- Custom chat clients can be used with `ChatAgent` to leverage all agent framework features +- Use the `create_agent()` method to easily create agents from your custom chat clients + +Both approaches allow you to extend the framework for your specific use cases while maintaining compatibility with the broader Agent Framework ecosystem. \ No newline at end of file diff --git a/python/samples/getting_started/agents/custom/custom_agent.py b/python/samples/getting_started/agents/custom/custom_agent.py new file mode 100644 index 0000000000..045dce2802 --- /dev/null +++ b/python/samples/getting_started/agents/custom/custom_agent.py @@ -0,0 +1,215 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +from collections.abc import AsyncIterable +from typing import Any + +from agent_framework import ( + AgentRunResponse, + AgentRunResponseUpdate, + AgentThread, + BaseAgent, + ChatMessage, + Role, + TextContent, +) + +""" +Custom Agent Implementation Example + +This sample demonstrates how to implement a custom agent by extending the BaseAgent class. +Custom agents provide complete control over the agent's behavior and capabilities, allowing +developers to create specialized agents that don't rely on chat clients. + +This approach is useful when you need to: +- Implement agents with custom logic that doesn't involve LLM interactions +- Create agents that integrate with specialized APIs or services +- Build agents with deterministic behaviors +- Implement new agent types for the Microsoft Agent Framework + +The EchoAgent example shows the minimal requirements for implementing a custom agent, +including both streaming and non-streaming response handling. +""" + + +class EchoAgent(BaseAgent): + """A simple custom agent that echoes user messages with a prefix. + + This demonstrates how to create a fully custom agent by extending BaseAgent + and implementing the required run() and run_stream() methods. + """ + + echo_prefix: str = "Echo: " + + def __init__( + self, + *, + name: str | None = None, + description: str | None = None, + echo_prefix: str = "Echo: ", + **kwargs: Any, + ) -> None: + """Initialize the EchoAgent. + + Args: + name: The name of the agent. + description: The description of the agent. + echo_prefix: The prefix to add to echoed messages. + **kwargs: Additional keyword arguments passed to BaseAgent. + """ + super().__init__( + name=name, + description=description, + echo_prefix=echo_prefix, # type: ignore + **kwargs, + ) + + async def run( + self, + messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None, + *, + thread: AgentThread | None = None, + **kwargs: Any, + ) -> AgentRunResponse: + """Execute the agent and return a complete response. + + Args: + messages: The message(s) to process. + thread: The conversation thread (optional). + **kwargs: Additional keyword arguments. + + Returns: + An AgentRunResponse containing the agent's reply. + """ + # Normalize input messages to a list + normalized_messages = self._normalize_messages(messages) + + if not normalized_messages: + response_message = ChatMessage( + role=Role.ASSISTANT, + contents=[TextContent(text="Hello! I'm a custom echo agent. Send me a message and I'll echo it back.")], + ) + else: + # For simplicity, echo the last user message + last_message = normalized_messages[-1] + if last_message.text: + echo_text = f"{self.echo_prefix}{last_message.text}" + else: + echo_text = f"{self.echo_prefix}[Non-text message received]" + + response_message = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text=echo_text)]) + + # Notify the thread of new messages if provided + if thread is not None: + await self._notify_thread_of_new_messages(thread, normalized_messages) + await self._notify_thread_of_new_messages(thread, response_message) + + return AgentRunResponse(messages=[response_message]) + + async def run_stream( + self, + messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None, + *, + thread: AgentThread | None = None, + **kwargs: Any, + ) -> AsyncIterable[AgentRunResponseUpdate]: + """Execute the agent and yield streaming response updates. + + Args: + messages: The message(s) to process. + thread: The conversation thread (optional). + **kwargs: Additional keyword arguments. + + Yields: + AgentRunResponseUpdate objects containing chunks of the response. + """ + # Normalize input messages to a list + normalized_messages = self._normalize_messages(messages) + + if not normalized_messages: + response_text = "Hello! I'm a custom echo agent. Send me a message and I'll echo it back." + else: + # For simplicity, echo the last user message + last_message = normalized_messages[-1] + if last_message.text: + response_text = f"{self.echo_prefix}{last_message.text}" + else: + response_text = f"{self.echo_prefix}[Non-text message received]" + + # Notify the thread of input messages if provided + if thread is not None: + await self._notify_thread_of_new_messages(thread, normalized_messages) + + # Simulate streaming by yielding the response word by word + words = response_text.split() + for i, word in enumerate(words): + # Add space before word except for the first one + chunk_text = f" {word}" if i > 0 else word + + yield AgentRunResponseUpdate( + contents=[TextContent(text=chunk_text)], + role=Role.ASSISTANT, + ) + + # Small delay to simulate streaming + await asyncio.sleep(0.1) + + # Notify the thread of the complete response if provided + if thread is not None: + complete_response = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text=response_text)]) + await self._notify_thread_of_new_messages(thread, complete_response) + + +async def main() -> None: + """Demonstrates how to use the custom EchoAgent.""" + print("=== Custom Agent Example ===\n") + + # Create EchoAgent + print("--- EchoAgent Example ---") + echo_agent = EchoAgent( + name="EchoBot", description="A simple agent that echoes messages with a prefix", echo_prefix="🔊 Echo: " + ) + + # Test non-streaming + print(f"Agent Name: {echo_agent.name}") + print(f"Agent ID: {echo_agent.id}") + print(f"Display Name: {echo_agent.display_name}") + + query = "Hello, custom agent!" + print(f"\nUser: {query}") + result = await echo_agent.run(query) + print(f"Agent: {result.messages[0].text}") + + # Test streaming + query2 = "This is a streaming test" + print(f"\nUser: {query2}") + print("Agent: ", end="", flush=True) + async for chunk in echo_agent.run_stream(query2): + if chunk.text: + print(chunk.text, end="", flush=True) + print() + + # Example with threads + print("\n--- Using Custom Agent with Thread ---") + thread = echo_agent.get_new_thread() + + # First message + result1 = await echo_agent.run("First message", thread=thread) + print("User: First message") + print(f"Agent: {result1.messages[0].text}") + + # Second message in same thread + result2 = await echo_agent.run("Second message", thread=thread) + print("User: Second message") + print(f"Agent: {result2.messages[0].text}") + + # Check conversation history + if thread.message_store: + messages = await thread.message_store.list_messages() + print(f"\nThread contains {len(messages)} messages in history") + else: + print("\nThread has no message store configured") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/samples/getting_started/agents/custom/custom_chat_client.py b/python/samples/getting_started/agents/custom/custom_chat_client.py new file mode 100644 index 0000000000..b138413194 --- /dev/null +++ b/python/samples/getting_started/agents/custom/custom_chat_client.py @@ -0,0 +1,183 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +import random +from collections.abc import AsyncIterable, MutableSequence +from typing import Any + +from agent_framework import ( + BaseChatClient, + ChatMessage, + ChatOptions, + ChatResponse, + ChatResponseUpdate, + Role, + TextContent, + use_function_invocation, +) + +""" +Custom Chat Client Implementation Example + +This sample demonstrates how to implement a custom chat client by extending the BaseChatClient class. +Custom chat clients allow you to integrate any backend service or create new LLM providers +for the Microsoft Agent Framework. + +This approach is useful when you need to: +- Integrate with new or proprietary LLM services +- Create mock implementations for testing +- Add custom authentication or routing logic +- Implement specialized preprocessing or postprocessing of requests and responses +- Create new LLM providers that work seamlessly with the framework's ChatAgent + +The EchoingChatClient example shows the minimal requirements for implementing a custom chat client, +including both streaming and non-streaming response handling, and demonstrates how to use the +custom client with ChatAgent through the create_agent() method. +""" + + +@use_function_invocation +class EchoingChatClient(BaseChatClient): + """A custom chat client that echoes messages back with modifications. + + This demonstrates how to implement a custom chat client by extending BaseChatClient + and implementing the required _inner_get_response() and _inner_get_streaming_response() methods. + """ + + OTEL_PROVIDER_NAME: str = "EchoingChatClient" + + prefix: str = "Echo:" + + def __init__(self, *, prefix: str = "Echo:", **kwargs: Any) -> None: + """Initialize the EchoingChatClient. + + Args: + prefix: Prefix to add to echoed messages. + **kwargs: Additional keyword arguments passed to BaseChatClient. + """ + super().__init__( + prefix=prefix, # type: ignore + **kwargs, + ) + + async def _inner_get_response( + self, + *, + messages: MutableSequence[ChatMessage], + chat_options: ChatOptions, + **kwargs: Any, + ) -> ChatResponse: + """Echo back the user's message with a prefix.""" + if not messages: + response_text = "No messages to echo!" + else: + # Echo the last user message + last_user_message = None + for message in reversed(messages): + if message.role == Role.USER: + last_user_message = message + break + + if last_user_message and last_user_message.text: + response_text = f"{self.prefix} {last_user_message.text}" + else: + response_text = f"{self.prefix} [No text message found]" + + response_message = ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text=response_text)]) + + return ChatResponse( + messages=[response_message], + model_id="echo-model-v1", + response_id=f"echo-resp-{random.randint(1000, 9999)}", + ) + + async def _inner_get_streaming_response( + self, + *, + messages: MutableSequence[ChatMessage], + chat_options: ChatOptions, + **kwargs: Any, + ) -> AsyncIterable[ChatResponseUpdate]: + """Stream back the echoed message character by character.""" + # Get the complete response first + response = await self._inner_get_response(messages=messages, chat_options=chat_options, **kwargs) + + if response.messages: + response_text = response.messages[0].text or "" + + # Stream character by character + for char in response_text: + yield ChatResponseUpdate( + contents=[TextContent(text=char)], + role=Role.ASSISTANT, + response_id=f"echo-stream-resp-{random.randint(1000, 9999)}", + ai_model_id="echo-model-v1", + ) + await asyncio.sleep(0.05) + + +async def main() -> None: + """Demonstrates how to implement and use a custom chat client with ChatAgent.""" + print("=== Custom Chat Client Example ===\n") + + # Create the custom chat client + print("--- EchoingChatClient Example ---") + + echo_client = EchoingChatClient(prefix="🔊 Echo:") + + # Use the chat client directly + print("Using chat client directly:") + direct_response = await echo_client.get_response("Hello, custom chat client!") + print(f"Direct response: {direct_response.messages[0].text}") + + # Create an agent using the custom chat client + echo_agent = echo_client.create_agent( + name="EchoAgent", + instructions="You are a helpful assistant that echoes back what users say.", + ) + + print(f"\nAgent Name: {echo_agent.name}") + print(f"Agent Display Name: {echo_agent.display_name}") + + # Test non-streaming with agent + query = "This is a test message" + print(f"\nUser: {query}") + result = await echo_agent.run(query) + print(f"Agent: {result.messages[0].text}") + + # Test streaming with agent + query2 = "Stream this message back to me" + print(f"\nUser: {query2}") + print("Agent: ", end="", flush=True) + async for chunk in echo_agent.run_stream(query2): + if chunk.text: + print(chunk.text, end="", flush=True) + print() + + # Example: Using with threads and conversation history + print("\n--- Using Custom Chat Client with Thread ---") + + thread = echo_agent.get_new_thread() + + # Multiple messages in conversation + messages = [ + "Hello, I'm starting a conversation", + "How are you doing?", + "Thanks for chatting!", + ] + + for msg in messages: + result = await echo_agent.run(msg, thread=thread) + print(f"User: {msg}") + print(f"Agent: {result.messages[0].text}\n") + + # Check conversation history + if thread.message_store: + thread_messages = await thread.message_store.list_messages() + print(f"Thread contains {len(thread_messages)} messages") + else: + print("Thread has no message store configured") + + +if __name__ == "__main__": + asyncio.run(main())