mirror of
https://github.com/msoedov/agentic_security.git
synced 2026-06-24 06:09:55 +02:00
feat(add mcp client):
This commit is contained in:
@@ -0,0 +1,52 @@
|
||||
from mcp import ClientSession, StdioServerParameters, types
|
||||
from mcp.client.stdio import stdio_client
|
||||
import asyncio
|
||||
|
||||
# Create server parameters for stdio connection
|
||||
server_params = StdioServerParameters(
|
||||
command="python", # Executable
|
||||
args=["agentic_security/mcp/main.py"], # Your server script
|
||||
env=None, # Optional environment variables
|
||||
)
|
||||
|
||||
|
||||
async def run():
|
||||
async with stdio_client(server_params) as (read, write):
|
||||
async with ClientSession(read, write) as session:
|
||||
# Initialize the connection
|
||||
await session.initialize()
|
||||
|
||||
# List available prompts, resources, and tools
|
||||
prompts = await session.list_prompts()
|
||||
print(f"Available prompts: {prompts}")
|
||||
|
||||
resources = await session.list_resources()
|
||||
print(f"Available resources: {resources}")
|
||||
|
||||
tools = await session.list_tools()
|
||||
print(f"Available tools: {tools}")
|
||||
|
||||
# Call the echo tool
|
||||
echo_result = await session.call_tool(
|
||||
"echo_tool", arguments={"message": "Hello from client!"}
|
||||
)
|
||||
print(f"Tool result: {echo_result}")
|
||||
|
||||
# # Read the echo resource
|
||||
# echo_content, mime_type = await session.read_resource(
|
||||
# "echo://Hello_resource"
|
||||
# )
|
||||
# print(f"Resource content: {echo_content}")
|
||||
# print(f"Resource MIME type: {mime_type}")
|
||||
|
||||
# # Get and use the echo prompt
|
||||
# prompt_result = await session.get_prompt(
|
||||
# "echo_prompt", arguments={"message": "Hello prompt!"}
|
||||
# )
|
||||
# print(f"Prompt result: {prompt_result}")
|
||||
|
||||
# You can perform additional operations here as needed
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(run())
|
||||
Reference in New Issue
Block a user