MCP ile Private Agentic Akışları

curl http://localhost:1234/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "qwen2.5-7b-instruct",
    "messages": [
      {"role": "user", "content": "Merhaba! Sen kimsin?"}
    ],
    "temperature": 0.7
  }'

import requests
 
response = requests.post(
    "http://localhost:1234/v1/chat/completions",
    json={
        "model": "qwen2.5-7b-instruct",
        "messages": [
            {"role": "user", "content": "Private Agent ne demek? Kısaca açıkla."}
        ]
    }
)
 
print(response.json()["choices"][0]["message"]["content"])

mkdir -p ~/Documents/privateAgents/lab
cd ~/Documents/privateAgents/lab
python3 -m venv .venv
source .venv/bin/activate
pip install requests

New-Item -ItemType Directory -Force -Path "$env:USERPROFILE\Documents\privateAgents\lab"
Set-Location "$env:USERPROFILE\Documents\privateAgents\lab"
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install requests

"""
Private Agent Lab - Simple Tool Calling Agent
LM Studio + Python ile local tool calling örneği
"""
 
import json
import os
import requests
 
LM_STUDIO_URL = "http://localhost:1234/v1/chat/completions"
MODEL_NAME = "qwen2.5-7b-instruct"
 
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "list_directory",
            "description": "List files and folders in a directory",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "The directory path to list"
                    }
                },
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Read the contents of a file",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "The file path to read"
                    }
                },
                "required": ["path"]
            }
        }
    }
]
 
def list_directory(path: str) -> str:
    """List files in a directory"""
    try:
        items = os.listdir(path)
        result = []
        for item in items:
            full_path = os.path.join(path, item)
            item_type = "📁" if os.path.isdir(full_path) else "📄"
            result.append(f"{item_type} {item}")
        return "\n".join(result) if result else "Empty directory"
    except Exception as e:
        return f"Error: {str(e)}"
 
def read_file(path: str) -> str:
    """Read a file's contents"""
    try:
        with open(path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        return f"Error: {str(e)}"
 
TOOL_FUNCTIONS = {
    "list_directory": list_directory,
    "read_file": read_file
}
 
def call_llm(messages: list, tools: list = None) -> dict:
    """Call LM Studio API"""
    payload = {
        "model": MODEL_NAME,
        "messages": messages,
        "temperature": 0.7,
    }
    if tools:
        payload["tools"] = tools
        payload["tool_choice"] = "auto"
    response = requests.post(LM_STUDIO_URL, json=payload, timeout=60)
    return response.json()
 
def run_agent(user_message: str) -> str:
    """Run the agent with a user message"""
    print(f"\n👤 User: {user_message}\n")
    
    messages = [
        {
            "role": "system",
            "content": "You are a helpful assistant with access to filesystem tools. "
                       "Use tools when the user asks about files or directories."
        },
        {"role": "user", "content": user_message}
    ]
    
    print("🤖 Calling LLM...")
    response = call_llm(messages, TOOLS)
    assistant_message = response["choices"][0]["message"]
    
    if "tool_calls" in assistant_message and assistant_message["tool_calls"]:
        print(f"🔧 Tool calls detected!\n")
        messages.append(assistant_message)
        
        for tool_call in assistant_message["tool_calls"]:
            tool_name = tool_call["function"]["name"]
            arguments = json.loads(tool_call["function"]["arguments"])
            
            print(f"   📞 Tool: {tool_name}")
            print(f"   📋 Args: {arguments}")
            
            result = TOOL_FUNCTIONS[tool_name](**arguments)
            print(f"   ✅ Result: {result[:100]}...\n" if len(result) > 100 else f"   ✅ Result: {result}\n")
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call["id"],
                "name": tool_name,
                "content": result
            })
        
        final_response = call_llm(messages)
        return final_response["choices"][0]["message"]["content"]
    
    return assistant_message.get("content", "No response")
 
if __name__ == "__main__":
    print("🔒 Private Agent - Tool Calling Demo")
    print("="*40)
    result = run_agent("List the files in the current directory")
    print(f"🤖 Assistant: {result}")

python simple_agent.py

🔒 Private Agent - Tool Calling Demo
========================================
 
👤 User: List the files in the current directory
 
🤖 Calling LLM...
🔧 Tool calls detected!
 
   📞 Tool: list_directory
   📋 Args: {'path': '.'}
   ✅ Result: 📄 simple_agent.py
 
🤖 Assistant: The current directory contains:
📄 simple_agent.py

"""
Private Agent Lab - MCP Client Agent
MCP protokolü ile tool calling örneği
"""
 
import json
import subprocess
import requests
 
LM_STUDIO_URL = "http://localhost:1234/v1/chat/completions"
MODEL_NAME = "qwen2.5-7b-instruct"
 
class MCPClient:
    """Simple MCP Client that communicates with MCP servers via stdio"""
    
    def __init__(self, server_command: list[str]):
        self.server_command = server_command
        self.process = None
        self.request_id = 0
        self.tools = []
        
    def start(self):
        """Start the MCP server process"""
        print(f"🚀 Starting MCP server: {' '.join(self.server_command)}")
        self.process = subprocess.Popen(
            self.server_command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1
        )
        self._initialize()
        
    def _send_request(self, method: str, params: dict = None) -> dict:
        """Send JSON-RPC request to MCP server"""
        self.request_id += 1
        request = {
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": method,
        }
        if params:
            request["params"] = params
            
        request_str = json.dumps(request) + "\n"
        self.process.stdin.write(request_str)
        self.process.stdin.flush()
        
        response_str = self.process.stdout.readline()
        return json.loads(response_str) if response_str else {}
    
    def _initialize(self):
        """Initialize MCP connection"""
        response = self._send_request("initialize", {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "private-agent-lab", "version": "1.0.0"}
        })
        
        if "result" in response:
            server_name = response['result'].get('serverInfo', {}).get('name', 'unknown')
            print(f"✅ MCP initialized: {server_name}")
        
        self._send_request("notifications/initialized")
        self._list_tools()
        
    def _list_tools(self):
        """Get available tools from MCP server"""
        response = self._send_request("tools/list")
        if "result" in response:
            self.tools = response["result"].get("tools", [])
            print(f"🔧 MCP Tools: {[t['name'] for t in self.tools]}")
    
    def call_tool(self, name: str, arguments: dict) -> str:
        """Call an MCP tool via JSON-RPC"""
        response = self._send_request("tools/call", {
            "name": name,
            "arguments": arguments
        })
        
        if "result" in response:
            content = response["result"].get("content", [])
            if content:
                return content[0].get("text", str(content))
        return "No result"
    
    def stop(self):
        if self.process:
            self.process.terminate()
 
def main():
    import os
    
    print("\n" + "="*60)
    print("🔒 Private Agent Lab - MCP Client Agent")
    print("="*60)
    
    allowed_dir = os.path.expanduser("~/Documents")
    print(f"📁 Allowed directory: {allowed_dir}\n")
    
    client = MCPClient([
        "npx", "-y", "@modelcontextprotocol/server-filesystem", 
        allowed_dir
    ])
    
    try:
        client.start()
        
        print("\n📂 Listing directory via MCP...")
        result = client.call_tool("list_directory", {"path": allowed_dir})
        print(f"Result: {result[:200]}...")
        
    except FileNotFoundError:
        print("❌ npx not found! Install Node.js: https://nodejs.org/")
    finally:
        client.stop()
 
if __name__ == "__main__":
    main()

node --version  # v18+ gerekli

python mcp_agent.py

{
    "type": "function",
    "function": {
        "name": "search_notes",
        "description": "Search through saved notes and return matching content",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query"
                }
            },
            "required": ["query"]
        }
    }
}

NOTES_STORE = {
    "project": "Private Agent Eğitimi - AI geliştiriciler için 2 saatlik modül",
    "tools": "list_directory, read_file, write_file, search_notes",
    "status": "Geliştirme aşamasında, Lab 3 tamamlandı"
}
 
def search_notes(query: str) -> str:
    """Search through notes"""
    results = []
    query_lower = query.lower()
    for key, value in NOTES_STORE.items():
        if query_lower in key.lower() or query_lower in value.lower():
            results.append(f"📝 {key}: {value}")
    return "\n".join(results) if results else "No matching notes found"
 
# TOOL_FUNCTIONS'a ekleyin:
TOOL_FUNCTIONS["search_notes"] = search_notes

"""
Private Agent Lab - Interactive File Analyzer
"""
 
import json
import os
import requests
from datetime import datetime
 
LM_STUDIO_URL = "http://localhost:1234/v1/chat/completions"
MODEL_NAME = "qwen2.5-7b-instruct"
 
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "list_directory",
            "description": "List files and folders in a directory",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string", "description": "Directory path"}},
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Read contents of a file",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string", "description": "File path"}},
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_file_info",
            "description": "Get detailed information about a file",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string", "description": "File path"}},
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "analyze_code",
            "description": "Analyze a Python file and return statistics",
            "parameters": {
                "type": "object",
                "properties": {"path": {"type": "string", "description": "Python file path"}},
                "required": ["path"]
            }
        }
    }
]
 
def list_directory(path: str) -> str:
    try:
        items = os.listdir(path)
        result = []
        for item in items:
            full_path = os.path.join(path, item)
            icon = "📁" if os.path.isdir(full_path) else "📄"
            result.append(f"{icon} {item}")
        return "\n".join(result) if result else "Empty directory"
    except Exception as e:
        return f"Error: {str(e)}"
 
def read_file(path: str) -> str:
    try:
        with open(path, 'r', encoding='utf-8') as f:
            return f.read()
    except Exception as e:
        return f"Error: {str(e)}"
 
def get_file_info(path: str) -> str:
    try:
        stat = os.stat(path)
        size = stat.st_size
        modified = datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M')
        return f"Size: {size} bytes\nModified: {modified}\nPath: {os.path.abspath(path)}"
    except Exception as e:
        return f"Error: {str(e)}"
 
def analyze_code(path: str) -> str:
    try:
        with open(path, 'r', encoding='utf-8') as f:
            content = f.read()
        lines = content.split('\n')
        functions = [l for l in lines if l.strip().startswith('def ')]
        classes = [l for l in lines if l.strip().startswith('class ')]
        imports = [l for l in lines if l.strip().startswith('import ') or l.strip().startswith('from ')]
        return f"""📊 Code Analysis:
- Total lines: {len(lines)}
- Functions: {len(functions)}
- Classes: {len(classes)}
- Imports: {len(imports)}
- Comments: {len([l for l in lines if l.strip().startswith('#')])}"""
    except Exception as e:
        return f"Error: {str(e)}"
 
TOOL_FUNCTIONS = {
    "list_directory": list_directory,
    "read_file": read_file,
    "get_file_info": get_file_info,
    "analyze_code": analyze_code
}
 
def call_llm(messages, tools=None):
    payload = {"model": MODEL_NAME, "messages": messages, "temperature": 0.7}
    if tools:
        payload["tools"] = tools
        payload["tool_choice"] = "auto"
    return requests.post(LM_STUDIO_URL, json=payload, timeout=60).json()
 
def run_agent(user_message):
    messages = [
        {"role": "system", "content": "You are a file system assistant. Use tools to help users explore and analyze files."},
        {"role": "user", "content": user_message}
    ]
    
    response = call_llm(messages, TOOLS)
    assistant_message = response["choices"][0]["message"]
    
    if "tool_calls" in assistant_message and assistant_message["tool_calls"]:
        messages.append(assistant_message)
        for tc in assistant_message["tool_calls"]:
            name = tc["function"]["name"]
            args = json.loads(tc["function"]["arguments"])
            print(f"  🔧 {name}({args})")
            result = TOOL_FUNCTIONS[name](**args)
            messages.append({"role": "tool", "tool_call_id": tc["id"], "name": name, "content": result})
        final = call_llm(messages)
        return final["choices"][0]["message"]["content"]
    return assistant_message.get("content", "")
 
if __name__ == "__main__":
    print("\n🔒 Private File Analyzer Agent")
    print("="*40)
    print("Commands: 'quit' to exit\n")
    
    while True:
        user_input = input("👤 You: ").strip()
        if user_input.lower() in ['quit', 'exit', 'q']:
            break
        if user_input:
            print()
            response = run_agent(user_input)
            print(f"\n🤖 Assistant: {response}\n")

List files in the current directory
Analyze the simple_agent.py file  
Get info about simple_agent.py