agent/agent-framework
2
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: Add custom agent and chat client implementation examples (#849)

Browse Source 
* 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>
This commit is contained in:
Copilot
2025-09-21 21:42:33 -07:00
committed by GitHub
Unverified
parent f61d8abe58
commit 39566d1dbf
3 changed files with 424 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/samples/getting_started/agents/custom/README.md
+26
View File
@@ -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.
python/samples/getting_started/agents/custom/custom_agent.py
+215
View File
@@ -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())
python/samples/getting_started/agents/custom/custom_chat_client.py
+183
View File
@@ -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())