server.py•6.61 kB
import asyncio
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
from pydantic import AnyUrl
import mcp.server.stdio
from src.agents.agents import script_using_topic
# Store notes as a simple key-value dict to demonstrate state management
notes: dict[str, str] = {}
script_server = Server("script_generator_server")
async def script_generate(topic: str, keywords: str) -> str:
task = f"""topic is: {topic}.
keywords is : {keywords} use these keyword as reference to generate script
where
Time: This likely refers to the duration or length of the script.
Objective: This describes the main goal or purpose of the script.
Audience: This specifies who the script is intended for.
Gender: This might refer to the gender of the target audience or the gender of the characters/speakers in the script.
Tone: This describes the overall mood or attitude of the script.
Speakers: This likely refers to the characters in the script."""
print("task", task)
re = await script_using_topic(task)
return re
@script_server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
"""
List available note resources.
Each note is exposed as a resource with a custom note:// URI scheme.
"""
return [
types.Resource(
uri=AnyUrl(f"note://internal/{name}"),
name=f"Note: {name}",
description=f"A simple note named {name}",
mimeType="text/plain",
)
for name in notes
]
@script_server.read_resource()
async def handle_read_resource(uri: AnyUrl) -> str:
"""
Read a specific note's content by its URI.
The note name is extracted from the URI host component.
"""
if uri.scheme != "note":
raise ValueError(f"Unsupported URI scheme: {uri.scheme}")
name = uri.path
if name is not None:
name = name.lstrip("/")
return notes[name]
raise ValueError(f"Note not found: {name}")
@script_server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
"""
List available prompts.
Each prompt can have optional arguments to customize its behavior.
"""
return [
types.Prompt(
name="summarize-notes",
description="Creates a summary of all notes",
arguments=[
types.PromptArgument(
name="style",
description="Style of the summary (brief/detailed)",
required=False,
)
],
)
]
@script_server.get_prompt()
async def handle_get_prompt(
name: str, arguments: dict[str, str] | None
) -> types.GetPromptResult:
"""
Generate a prompt by combining arguments with server state.
The prompt includes all current notes and can be customized via arguments.
"""
if name != "summarize-notes":
raise ValueError(f"Unknown prompt: {name}")
style = (arguments or {}).get("style", "brief")
detail_prompt = " Give extensive details." if style == "detailed" else ""
return types.GetPromptResult(
description="Summarize the current notes",
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text=f"Here are the current notes to summarize:{detail_prompt}\n\n"
+ "\n".join(
f"- {name}: {content}" for name, content in notes.items()
),
),
)
],
)
@script_server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
List available tools.
Each tool specifies its arguments using JSON Schema validation.
"""
return [
types.Tool(
name="script_generate",
description="Provide topic and keyword to generat Script",
inputSchema={
"type": "object",
"properties": {
"topic": {"type": "string"},
"keywords": {"type": "string"},
},
"required": ["topic", "keywords"],
},
)
]
@script_server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""
Handle tool execution requests.
Tools can modify server state and notify clients of changes.
"""
if name != "script_using_topic":
raise ValueError(f"Unknown tool: {name}")
if not arguments:
raise ValueError("Missing arguments")
if name == "script_using_topic":
try:
topic = arguments.get("topic")
keywords = arguments.get("keywords")
response = await script_generate(topic=topic, keywords=keywords)
if isinstance(response, bytes):
response = response.decode("utf-8")
p = [
types.TextContent(
type="text",
text=response,
)
]
print(f"[RESPONSE] {p}")
return [
types.TextContent(
type="text",
text=f'{{"script": "{response}"}}',
)
]
except Exception as e:
print(f"[Error from script_using_topic]{e}")
return [
types.TextContent(
type="text",
text=f"Error: {e}",
)
]
async def main():
# Run the server using stdin/stdout streams
async with mcp.script_server.stdio.stdio_server() as (read_stream, write_stream):
await script_server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="script_generator_server",
server_version="0.1.0",
capabilities=script_server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
# Entrypoint
if __name__ == "__main__":
# Run the server
asyncio.run(main())