import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from 'express';
import { z } from 'zod';
import { CallToolRequestSchema, ListToolsRequestSchema, } from "@modelcontextprotocol/sdk/types.js";
const { Request, Response } = express;
const GetArgumentsSchema = z.object({
  key: z.string(),
});
// 模擬 redis
const memory = {"hello": "world"};
// 創(chuàng)建 MCP Server
const server = new Server({
  name: "example-server",
  version: "1.0.0",
},{
  capabilities: {
    tools: {},
  }
});
// 注冊(cè)獲取工具 API 路由
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
      tools: [
          {
              name: "get",
              description: "Get value by key from kv",
              inputSchema: {
                  type: "object",
                  properties: {
                      key: {
                          type: "string",
                          description: "",
                      },
                  },
                  required: ["key"],
              },
          },
      ],
  };
});
// 注冊(cè)執(zhí)行工具 API 路由
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  console.log(request.params)
  const { name, arguments: args } = request.params;
  const { key } = GetArgumentsSchema.parse(args);
  return {
    content: [
        {
            type: "text",
            text: `${memory[key]}`,
        },
    ],
};
});
// 啟動(dòng) HTTP 服務(wù)器
const app = express();
const transports: {[sessionId: string]: SSEServerTransport} = {};
// SSE 路由
app.get("/sse", async (_: Request, res: Response) => {
  // 創(chuàng)建一個(gè) SSE 通道，meesages 為后續(xù)通信的路由
  const transport = new SSEServerTransport('/messages', res);
  transports[transport.sessionId] = transport;
  res.on("close", () => {
    delete transports[transport.sessionId];
  });
  await server.connect(transport);
});
// 通信路由
app.post("/messages", async (req: Request, res: Response) => {
  const sessionId = req.query.sessionId as string;
  const transport = transports[sessionId];
  if (transport) {
    // 處理請(qǐng)求，處理完畢后通過(guò)上面的 sse res 進(jìn)行響應(yīng)
    await transport.handlePostMessage(req, res);
  } else {
    res.status(400).send('No transport found for sessionId');
  }
});
app.listen(3001);

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";
async function runClient() {
    const client = new Client(
      {
        name: "mcp-typescript test client",
        version: "0.1.0",
      },
    );
  
  
    const clientTransport = new SSEClientTransport(new URL("http://localhost:3001/sse"));
  
    await client.connect(clientTransport);
    const tools = await client.listTools()  
    console.log(tools)
    const resp = await client.callTool({
      name: "get",
      arguments: {
        key: "hello"
      }
    })
    console.log(resp);
    await client.close();
  }
  runClient();

interface Transport {
  // 初始化通道
  start(): Promise<void>;
  // 發(fā)送信息或響應(yīng)
  send(message: JSONRPCMessage): Promise<void>;
  
  close(): Promise<void>;
  onclose?: () => void;
  onerror?: (error: Error) => void;
  // 處理收到的請(qǐng)求或響應(yīng)
  onmessage?: (message: JSONRPCMessage) => void;
  // 連接對(duì)應(yīng)的回話 ID
  sessionId?: string;
}

class InMemoryTransport implements Transport {
  private _otherTransport?: InMemoryTransport;
  private _messageQueue: JSONRPCMessage[] = [];
  onclose?: () => void;
  onerror?: (error: Error) => void;
  onmessage?: (message: JSONRPCMessage) => void;
  sessionId?: string;
  static createLinkedPair(): [InMemoryTransport, InMemoryTransport] {
    // 創(chuàng)建兩個(gè)管道，互相關(guān)聯(lián)
    const clientTransport = new InMemoryTransport();
    const serverTransport = new InMemoryTransport();
    clientTransport._otherTransport = serverTransport;
    serverTransport._otherTransport = clientTransport;
    return [clientTransport, serverTransport];
  }
  // 初始化管道，如果之前已經(jīng)存在數(shù)據(jù)，則消費(fèi)
  async start(): Promise<void> {
    while (this._messageQueue.length > 0) {
      const message = this._messageQueue.shift();
      if (message) {
        this.onmessage?.(message);
      }
    }
  }
  async close(): Promise<void> {
    const other = this._otherTransport;
    this._otherTransport = undefined;
    await other?.close();
    this.onclose?.();
  }
  // 發(fā)送數(shù)據(jù)
  async send(message: JSONRPCMessage): Promise<void> {
    if (!this._otherTransport) {
      throw new Error("Not connected");
    }
    // 如果還沒(méi)有消費(fèi)者，則先緩存
    if (this._otherTransport.onmessage) {
      this._otherTransport.onmessage(message);
    } else {
      this._otherTransport._messageQueue.push(message);
    }
  }
}

class SSEServerTransport implements Transport {
  private _sseResponse?: ServerResponse;
  private _sessionId: string;
  onclose?: () => void;
  onerror?: (error: Error) => void;
  // 上層設(shè)置
  onmessage?: (message: JSONRPCMessage) => void;
 
  // 初始化時(shí)記錄 SSE 連接對(duì)應(yīng)的響應(yīng)對(duì)象，以及用于通信的 endpoint 路由
  constructor(
    private _endpoint: string,
    private res: ServerResponse,
  ) {
    this._sessionId = randomUUID();
  }
  async start(): Promise<void> {
    // 返回 text/event-stream 類型的響應(yīng)頭，表示這是一個(gè) SSE 連接，后續(xù)可以基于這個(gè)連接進(jìn)行數(shù)據(jù)推送
    this.res.writeHead(200, {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
    });
    // 返回 endpoint 信息，作為 POST 請(qǐng)求的 url 路徑
    this.res.write(
      `event: endpoint\ndata: ${encodeURI(this._endpoint)}?sessinotallow=${this._sessionId}\n\n`,
    );
    // 記錄 SSE 連接對(duì)應(yīng)的響應(yīng)對(duì)象
    this._sseResponse = this.res;
  }
  // 處理 POST 請(qǐng)求，解析請(qǐng)求體，將請(qǐng)求體并通過(guò) SSE 響應(yīng)對(duì)象把請(qǐng)求結(jié)果發(fā)送給客戶端
  async handlePostMessage(
    req: IncomingMessage,
    res: ServerResponse,
    parsedBody?: unknown,
  ): Promise<void> {
    // 請(qǐng)求前置處理
    let body: string | unknown;
    try {
      const ct = contentType.parse(req.headers["content-type"] ?? "");
      if (ct.type !== "application/json") {
        throw new Error(`Unsupported content-type: ${ct}`);
      }
      body = parsedBody ?? await getRawBody(req, {
        limit: MAXIMUM_MESSAGE_SIZE,
        encoding: ct.parameters.charset ?? "utf-8",
      });
    } catch (error) {
      res.writeHead(400).end(String(error));
      this.onerror?.(error as Error);
      return;
    }
    try {
      // 具體的處理邏輯
      await this.handleMessage(typeof body === 'string' ? JSON.parse(body) : body);
    } catch {
      res.writeHead(400).end(`Invalid message: ${body}`);
      return;
    }
    // 先回復(fù) 202 Accepted，表示請(qǐng)求已經(jīng)被接受，后續(xù)處理邏輯由上層處理
    res.writeHead(202).end("Accepted");
  }
  async handleMessage(message: unknown): Promise<void> {
    let parsedMessage: JSONRPCMessage;
    // 解析請(qǐng)求，通知上層處理
    try {
      parsedMessage = JSONRPCMessageSchema.parse(message);
    } catch (error) {
      this.onerror?.(error as Error);
      throw error;
    }
    this.onmessage?.(parsedMessage);
  }
  async send(message: JSONRPCMessage): Promise<void> {
    this._sseResponse.write(
      `event: message\ndata: ${JSON.stringify(message)}\n\n`,
    );
  }
  get sessionId(): string {
    return this._sessionId;
  }
}

export class SSEClientTransport implements Transport {
  private _eventSource?: EventSource;
  private _endpoint?: URL;
  private _abortController?: AbortController;
  private _url: URL;
  private _eventSourceInit?: EventSourceInit;
  private _requestInit?: RequestInit;
  
  // 上層定義
  onmessage?: (message: JSONRPCMessage) => void;
  constructor(
    url: URL,
    opts?: SSEClientTransportOptions,
  ) {
    this._url = url;
    this._eventSourceInit = opts?.eventSourceInit;
    this._requestInit = opts?.requestInit;
    // 身份驗(yàn)證提供者
    this._authProvider = opts?.authProvider;
  }
  async start() {
    return await this._startOrAuth();
  }
  private _startOrAuth(): Promise<void> {
    return new Promise((resolve, reject) => {
      // SSE 客戶端
      this._eventSource = new EventSource(
        this._url.href,
        this._eventSourceInit ?? {
          // 自定義請(qǐng)求實(shí)現(xiàn)
          fetch: (url, init) => this._commonHeaders().then((headers) => fetch(url, {
            ...init,
            headers: {
              ...headers,
              Accept: "text/event-stream"
            }
          })),
        },
      );
      // 接收服務(wù)器的 endpoint，用于后續(xù)通信
      this._eventSource.addEventListener("endpoint", (event: Event) => {
        const messageEvent = event as MessageEvent;
        this._endpoint = new URL(messageEvent.data, this._url);
        resolve();
      });
      // 接收服務(wù)端消息
      this._eventSource.onmessage = (event: Event) => {
        const messageEvent = event as MessageEvent;
        let message: JSONRPCMessage;
        try {
          message = JSONRPCMessageSchema.parse(JSON.parse(messageEvent.data));
        } catch (error) {
          this.onerror?.(error as Error);
          return;
        }
        this.onmessage?.(message);
      };
    });
  }
  // 發(fā)送消息給客戶端
  async send(message: JSONRPCMessage): Promise<void> {
    const headers = new Headers({ ...this._requestInit?.headers });
    headers.set("content-type", "application/json");
    const init = {
      ...this._requestInit,
      method: "POST",
      headers,
      body: JSON.stringify(message),
      signal: this._abortController?.signal,
    };
    const response = await fetch(this._endpoint, init);
  }
}

class Server extends Protocol {
  private _clientCapabilities?: ClientCapabilities;
  private _clientVersion?: Implementation;
  private _capabilities: ServerCapabilities;
  private _instructions?: string;
  constructor(
    private _serverInfo: Implementation,
    options?: ServerOptions,
  ) {
    super(options);
    // 支持的能力
    this._capabilities = options?.capabilities ?? {};
    this._instructions = options?.instructions;
    // 注冊(cè)初始化請(qǐng)求路由和處理函數(shù)，該請(qǐng)求用戶獲取服務(wù)端的一些元信息，比如支持的能力
    this.setRequestHandler(InitializeRequestSchema, (request) =>
      this._oninitialize(request),
    );
  }
  private async _oninitialize(
    request: InitializeRequest,
  ): Promise<InitializeResult> {
    const requestedVersion = request.params.protocolVersion;
    this._clientCapabilities = request.params.capabilities;
    this._clientVersion = request.params.clientInfo;
    return {
      protocolVersion: SUPPORTED_PROTOCOL_VERSIONS.includes(requestedVersion)
        ? requestedVersion
        : LATEST_PROTOCOL_VERSION,
      capabilities: this.getCapabilities(),
      serverInfo: this._serverInfo,
      ...(this._instructions && { instructions: this._instructions }),
    };
  }
  // 注冊(cè) Server 支持的能力，比如工具、Prompt
  public registerCapabilities(capabilities: ServerCapabilities): void {
    this._capabilities = mergeCapabilities(this._capabilities, capabilities);
  }
}

export abstract class Protocol {
  private _transport?: Transport;
  private _requestMessageId = 0;
  private _requestHandlers = new Map();
  private _responseHandlers = new Map();
  // 初始化實(shí)例，保存 transport，后續(xù)用 transport 和對(duì)象通信
  async connect(transport: Transport): Promise<void> {
    this._transport = transport;
    this._transport.onmessage = (message) => {
      // 根據(jù) message 的類型，分別處理請(qǐng)求、響應(yīng)、通知，因?yàn)樵擃悤?huì)被 Clien 和 Server 繼承
      if (!("method" in message)) {
        this._onresponse(message);
      } else if ("id" in message) {
        this._onrequest(message);
      } else {
        this._onnotification(message);
      }
    };
    await this._transport.start();
  }
  private _onrequest(request: JSONRPCRequest): void {
    const handler = this._requestHandlers.get(request.method) ?? this.fallbackRequestHandler;
    const extra = {
      sessionId: this._transport?.sessionId,
    };
    Promise.resolve()
      .then(() => handler(request, extra)) // 處理請(qǐng)求
      .then(
        (result) => {
          // 發(fā)送處理請(qǐng)求的結(jié)果
          return this._transport?.send({
            result,
            jsonrpc: "2.0",
            id: request.id,
          });
        })
  }
  private _onresponse(response: JSONRPCResponse | JSONRPCError): void {
    const messageId = Number(response.id);
    const handler = this._responseHandlers.get(messageId);
    this._responseHandlers.delete(messageId);
    handler(response);
  }
  request<T extends ZodType<object>>(
    request: SendRequestT,
    resultSchema: T,
    options?: RequestOptions,
  ): Promise<z.infer<T>> {
    return new Promise((resolve, reject) => {
      // 請(qǐng)求 ID 遞增
      const messageId = this._requestMessageId++;
      const jsonrpcRequest: JSONRPCRequest = {
        ...request,
        jsonrpc: "2.0",
        id: messageId,
      };
      // 設(shè)置請(qǐng)求和處理函數(shù)的映射，在 _onresponse 處理響應(yīng)
      this._responseHandlers.set(messageId, (response) => {
        try {
          const result = resultSchema.parse(response.result);
          resolve(result);
        } catch (error) {
          reject(error);
        }
      });
      // 發(fā)送請(qǐng)求
      this._transport.send(jsonrpcRequest);
    });
  }
  // 注冊(cè)路由和處理函數(shù)
  setRequestHandler<
    T extends ZodObject<{
      method: ZodLiteral<string>;
    }>,
  >(
    requestSchema: T,
    handler: (
      request: z.infer<T>,
      extra: RequestHandlerExtra,
    ) => SendResultT | Promise<SendResultT>,
  ): void {
    const method = requestSchema.shape.method.value;
    // 存入 map 
    this._requestHandlers.set(method, (request, extra) =>
      Promise.resolve(handler(requestSchema.parse(request), extra)),
    );
  }
}

export class McpServer {
  public readonly server: Server;
  private _registeredTools = {};
  constructor(serverInfo: Implementation, options?: ServerOptions) {
  private _registeredTools: { [name: string]: Tool } = {};
  }
  async connect(transport: Transport): Promise<void> {
    return await this.server.connect(transport);
  }
  private setToolRequestHandlers() {
    // 表示 Server 支持工具
    this.server.registerCapabilities({
      tools: {},
    });
    // 注冊(cè)獲取工具列表路由
    this.server.setRequestHandler(
      ListToolsRequestSchema,
      (): ListToolsResult => ({
        tools: Object.entries(this._registeredTools).map(
          ([name, tool]): Tool => {
            return {
              name,
              description: tool.description,
              inputSchema: tool.inputSchema
                ? (zodToJsonSchema(tool.inputSchema, {
                    strictUnions: true,
                  }) as Tool["inputSchema"])
                : EMPTY_OBJECT_JSON_SCHEMA,
            };
          },
        ),
      }),
    );
    // 注冊(cè)調(diào)用工具路由
    this.server.setRequestHandler(
      CallToolRequestSchema,
      async (request, extra): Promise<CallToolResult> => {
        // 獲取對(duì)應(yīng)的工具
        const tool = this._registeredTools[request.params.name];
        const parseResult = await tool.inputSchema.safeParseAsync(
          request.params.arguments,
        );
        // 解析參數(shù)
        const args = parseResult.data;
        const cb = tool.callback as ToolCallback<ZodRawShape>;
        return await Promise.resolve(cb(args, extra));
      },
    );
  }
  // 注冊(cè)工具，name為工具名，description為工具描述，paramsSchema為參數(shù)的zod schema，cb為工具的回調(diào)函數(shù)
  tool(name: string, ...rest: unknown[]): void {
    if (this._registeredTools[name]) {
      throw new Error(`Tool ${name} is already registered`);
    }
    let description;
    if (typeof rest[0] === "string") {
      description = rest.shift() as string;
    }
    let paramsSchema;
    if (rest.length > 1) {
      paramsSchema = rest.shift() as ZodRawShape;
    }
    const cb = rest[0];
    // 記錄工具信息
    this._registeredTools[name] = {
      description,
      inputSchema:
        paramsSchema === undefined ? undefined : z.object(paramsSchema),
      callback: cb,
    };
    // 給 MCP Server 注冊(cè)路由，這樣就可以處理客戶端的工具相關(guān)請(qǐng)求
    this.setToolRequestHandlers();
  }
}

export class Client extends Protocol {
  private _serverCapabilities?: ServerCapabilities;
  private _serverVersion?: Implementation;
  private _capabilities: ClientCapabilities;
  private _instructions?: string;
  constructor(
    private _clientInfo: Implementation,
    options?: ClientOptions,
  ) {
    super(options);
    // 客戶端支持的能力
    this._capabilities = options?.capabilities ?? {};
  }
  public registerCapabilities(capabilities: ClientCapabilities): void {
    this._capabilities = mergeCapabilities(this._capabilities, capabilities);
  }
  override async connect(transport: Transport): Promise<void> {
    // 建立和服務(wù)器的連接和身份驗(yàn)證，由 Transport 實(shí)現(xiàn)提供
    await super.connect(transport);
    try {
      // 發(fā)送初始化請(qǐng)求獲取服務(wù)器元信息，比如支持的能力
      const result = await this.request(
        {
          method: "initialize",
          params: {
            protocolVersion: LATEST_PROTOCOL_VERSION,
            capabilities: this._capabilities,
            clientInfo: this._clientInfo,
          },
        },
        InitializeResultSchema,
      );
      if (result === undefined) {
        throw new Error(`Server sent invalid initialize result: ${result}`);
      }
      if (!SUPPORTED_PROTOCOL_VERSIONS.includes(result.protocolVersion)) {
        throw new Error(
          `Server's protocol version is not supported: ${result.protocolVersion}`,
        );
      }
      this._serverCapabilities = result.capabilities;
      this._serverVersion = result.serverInfo;
      this._instructions = result.instructions;
      await this.notification({
        method: "notifications/initialized",
      });
    } catch (error) {
      // Disconnect if initialization fails.
      void this.close();
      throw error;
    }
  }
  // 調(diào)用工具
  async callTool(
    params: CallToolRequest["params"],
    resultSchema,
    options?: RequestOptions,
  ) {
    return this.request(
      { method: "tools/call", params },
      resultSchema,
      options,
    );
  }
  // 獲取工具列表
  async listTools(
    params?: ListToolsRequest["params"],
    options?: RequestOptions,
  ) {
    return this.request(
      { method: "tools/list", params },
      ListToolsResultSchema,
      options,
    );
  }
}

async def run_mcp() -> None:
    """Main entry point for the MCP runner."""
    args = parse_args()
    runner = MCPRunner()
    await runner.initialize(args.connection, args.server_url)
    await runner.run_default()
if __name__ == "__main__":
    asyncio.run(run_mcp())

class MCPRunner:
    def __init__(self):
        self.agent = MCPAgent()
    async def initialize(
        self,
        connection_type: str,
        server_url: str | None = None,
    ) -> None:
        // 建立和 MCP 服務(wù)器的連接
        await self.agent.initialize(connection_type="sse", server_url=server_url)
    async def run_default(self) -> None:
        # 提示用戶輸入問(wèn)題
        prompt = input("Enter your prompt: ")
        # 開(kāi)始處理用戶輸入的問(wèn)題
        await self.agent.run(prompt)

class MCPAgent(ToolCallAgent):
    mcp_clients: MCPClients = Field(default_factory=MCPClients)
    
    async def initialize(
        self,
        connection_type: Optional[str] = None,
        server_url: Optional[str] = None,
        command: Optional[str] = None,
        args: Optional[List[str]] = None,
    ) -> None:
        
        # 初始化和 MCP 服務(wù)器的連接，并獲取工具列表
        await self.mcp_clients.connect_sse(server_url=server_url)
        # 保存工具相關(guān)的實(shí)例到 available_tools
        self.available_tools = self.mcp_clients
    async def run(self, request: Optional[str] = None) -> str:
        result = await super().run(request)
        return result

class MCPClients(ToolCollection):
    session: Optional[ClientSession] = None
    exit_stack: AsyncExitStack = None
    description: str = "MCP client tools for server interaction"
    def __init__(self):
        super().__init__()  # Initialize with empty tools list
        self.name = "mcp"  # Keep name for backward compatibility
        self.exit_stack = AsyncExitStack()
    async def connect_sse(self, server_url: str) -> None:
        streams_context = sse_client(url=server_url)
        streams = await self.exit_stack.enter_async_context(streams_context)
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(*streams)
        )
        await self._initialize_and_list_tools()
    # 從 MCP Server 獲取工具列表
    async def _initialize_and_list_tools(self) -> None:
        
        await self.session.initialize()
        # 從 MCP Server 獲取工具列表
        response = await self.session.list_tools()
        # 存起來(lái)
        self.tools = tuple()
        self.tool_map = {}
        # Create proper tool objects for each server tool
        for tool in response.tools:
            server_tool = MCPClientTool(
                name=tool.name,
                descriptinotallow=tool.description,
                parameters=tool.inputSchema,
                sessinotallow=self.session,
            )
            self.tool_map[tool.name] = server_tool
        self.tools = tuple(self.tool_map.values())

async def run(self, request: Optional[str] = None) -> str:
        
        if self.state != AgentState.IDLE:
            raise RuntimeError(f"Cannot run agent from state: {self.state}")
        if request:
            self.update_memory("user", request)
        results: List[str] = []
        # 循環(huán)調(diào)大模型
        async with self.state_context(AgentState.RUNNING):
            while (
                self.current_step < self.max_steps and self.state != AgentState.FINISHED
            ):
                self.current_step += 1
                logger.info(f"Executing step {self.current_step}/{self.max_steps}")
                # 不斷調(diào) step，子類實(shí)現(xiàn)
                step_result = await self.step()
                results.append(f"Step {self.current_step}: {step_result}")
            if self.current_step >= self.max_steps:
                self.current_step = 0
                self.state = AgentState.IDLE
                results.append(f"Terminated: Reached max steps ({self.max_steps})")
        await SANDBOX_CLIENT.cleanup()
        return "\n".join(results) if results else "No steps executed"

async def step(self) -> str:
        """Execute a single step: think and act."""
        should_act = await self.think()
        if not should_act:
            return "Thinking complete - no action needed"
        return await self.act()

async def think(self) -> bool:
        
        try:
            # 傳入工具調(diào)大模型，詢問(wèn)需要執(zhí)行的工具
            response = await self.llm.ask_tool(
                messages=self.messages,
                # 傳入可用的工具
                tools=self.available_tools.to_params(),
                tool_choice=self.tool_choices,
            )
        
        self.tool_calls = tool_calls = (
            response.tool_calls if response and response.tool_calls else []
        )
    async def act(self) -> str:
        results = []
        # 執(zhí)行模型返回的工具列表
        for command in self.tool_calls:
            result = await self.execute_tool(command)
            results.append(result)
        return "\n\n".join(results)
    # 執(zhí)行某個(gè)工具
    async def execute_tool(self, command: ToolCall) -> str:
       
        name = command.function.name
        if name not in self.available_tools.tool_map:
            return f"Error: Unknown tool '{name}'"
        try:
            # 解析大模型提取的參數(shù)
            args = json.loads(command.function.arguments or "{}")
            # 執(zhí)行工具
            result = await self.available_tools.execute(name=name, tool_input=args)
            # Format result for display (standard case)
            observation = (
                f"Observed output of cmd `{name}` executed:\n{str(result)}"
                if result
                else f"Cmd `{name}` completed with no output"
            )
            return observation

class _BashSession:
    command: str = "/bin/bash"
    async def start(self):
        # 創(chuàng)建一個(gè) bash 進(jìn)程
        self._process = await asyncio.create_subprocess_shell(
            self.command,
            preexec_fn=os.setsid,
            shell=True,
            bufsize=0,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
    async def run(self, command: str):
        # 讓 bash 執(zhí)行命令
        self._process.stdin.write(
            command.encode() + f"; echo '{self._sentinel}'\n".encode()
        )
        await self._process.stdin.drain()
        # 獲取輸出
class Bash(BaseTool):
    """A tool for executing bash commands"""
    name: str = "bash"
    description: str = _BASH_DESCRIPTION
    parameters: dict = {
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "The bash command to execute. Can be empty to view additional logs when previous exit code is `-1`. Can be `ctrl+c` to interrupt the currently running process.",
            },
        },
        "required": ["command"],
    }
    _session: Optional[_BashSession] = None
    async def execute(
        self, command: str | None = None, restart: bool = False, **kwargs
    ) -> CLIResult:
        if restart:
            if self._session:
                self._session.stop()
            self._session = _BashSession()
            await self._session.start()
            return CLIResult(system="tool has been restarted.")
        if self._session is None:
            self._session = _BashSession()
            await self._session.start()
        if command is not None:
            return await self._session.run(command)
        raise ToolError("no command provided.")