test.py•2.65 kB
# test.py
import asyncio
import json
from mcp.client.sse import sse_client
from mcp import ClientSession
async def main():
# 1) Open the SSE connection in one async‐with
async with sse_client("http://127.0.0.1:8000/sse/") as (read_stream, write_stream):
# 2) Create the MCP client session over those streams
async with ClientSession(read_stream, write_stream) as session:
# 3) Initialize, then call the podcast generation tool
await session.initialize()
# Cat and dog podcast parameters
podcast_params = {
"prompt": "A humorous conversation between a cat and dog about their daily lives, their different perspectives on humans, and their secret adventures when their owners are away.",
"duration": 3, # 3-minute podcast
"name1": "Whiskers", # Cat's name
"voice1": "Dorothy", # Cat's voice
"first_person_quality1": "sophisticated",
"first_person_quality2": "snarky",
"gender1": "F", # Female cat
"age1": 5,
"name2": "Buddy", # Dog's name
"voice2": "Brian", # Dog's voice
"second_person_quality1": "enthusiastic",
"second_person_quality2": "loyal",
"gender2": "M", # Male dog
"age2": 4
}
print("Generating cat and dog podcast...")
result = await session.call_tool(
"generate_podcast",
podcast_params
)
# Pretty print the podcast information
print("\n=== Cat and Dog Podcast Generated ===")
print(f"Host: {podcast_params['name1']} (Cat)")
print(f"Guest: {podcast_params['name2']} (Dog)")
# Print the script with proper formatting
if isinstance(result, dict) and "script" in result:
print("\n=== Podcast Script ===")
print(result["script"])
# Print audio URLs if available
if isinstance(result, dict) and "audio_urls" in result:
print("\n=== Audio URLs ===")
for audio in result["audio_urls"]:
print(audio)
# Save the full result to a file for reference
with open("cat_dog_podcast_result.json", "w") as f:
json.dump(result, f, indent=2)
print("\nFull results saved to cat_dog_podcast_result.json")
if __name__ == "__main__":
asyncio.run(main())