semantic kernel - creating a mcp server and calling it using a sse client
To create a mcp server with semantic kernel we just have to declare a kernel_function like so
@kernel_function()
def echo_function(message: str, extra: str = "") -> str:
"""Echo a message as a function"""
return f"Hello hello function echo: {message} {extra}"
Integrate that into our kernel
kernel.add_function("echo", echo_function, "echo_function")
kernel.add_function(
plugin_name="prompt",
function_name="prompt",
prompt_template_config=PromptTemplateConfig(
name="prompt",
description="This is a prompt",
template="Please repeat this: {{$message}} and this: {{$extra}}",
input_variables=[
InputVariable(
name="message",
description="This is the message.",
is_required=True,
json_schema='{ "type": "string", "description": "This is the message."}',
),
InputVariable(
name="extra",
description="This is extra.",
default="default",
is_required=False,
json_schema='{ "type": "string", "description": "This is the message."}',
),
],
),
)
And then run our uvicorn on port 8000 via sse.
server = kernel.as_mcp_server(server_name="sk")
import uvicorn
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Mount, Route
sse = SseServerTransport("/messages/")
async def handle_sse(request):
async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
starlette_app = Starlette(
debug=True,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
)
config = uvicorn.Config(starlette_app, host="0.0.0.0", port=8000)
uvicorn_server = uvicorn.Server(config)
await uvicorn_server.serve()
Run our server
uv run main_mcp.py
The client to test this can be achieve via sse_client.
import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client
async def main():
async with sse_client("http://localhost:8000/sse") as (read, write):
async with ClientSession(read, write) as session:
# Initialize the session
await session.initialize()
# List available tools
tools = await session.list_tools()
print("Available tools:", tools.tools)
print()
# Call echo_function tool
result = await session.call_tool("echo_function", {"message": "Hello", "extra": "World"})
print("echo_function result:", result.content)
print()
if __name__ == "__main__":
asyncio.run(main())
And it should output something similiar to this here:-
Comments