✅ fix(tool): 支持 Claude Code Router 兼容 Claude Code

.env.example

@@ -5,16 +5,12 @@
 # API 认证配置
 # =============================================================================
 
-# 客户端认证密钥（OpenAI 和 Anthropic 共用）
+# 客户端认证密钥
 # 客户端调用时需要使用此密钥进行认证
 AUTH_TOKEN=sk-your-api-key
 # 是否跳过api key验证
 SKIP_AUTH_TOKEN=false
 
-# Anthropic API 客户端认证密钥（可选）
-# 如果未设置，将使用 AUTH_TOKEN 的值
-# ANTHROPIC_API_KEY=sk-your-api-key
-
 # 备用认证令牌（匿名模式失败时使用）
 BACKUP_TOKEN=eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjMxNmJjYjQ4LWZmMmYtNGExNS04NTNkLWYyYTI5YjY3ZmYwZiIsImVtYWlsIjoiR3Vlc3QtMTc1NTg0ODU4ODc4OEBndWVzdC5jb20ifQ.PktllDySS3trlyuFpTeIZf-7hl8Qu1qYF3BxjgIul0BrNux2nX9hVzIjthLXKMWAf9V0qM8Vm_iyDqkjPGsaiQ
 
@@ -38,6 +34,9 @@ THINKING_MODEL=GLM-4.5-Thinking
 # 搜索模式模型名称
 SEARCH_MODEL=GLM-4.5-Search
 
+# Air 模型名称
+AIR_MODEL=GLM-4.5-Air
+
 # =============================================================================
 # 服务器配置
 # =============================================================================

README.md

@@ -1,24 +1,24 @@
-# Z.AI OpenAI & Anthropic API 代理服务
+# Z.AI OpenAI API 代理服务
 
 ![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)
 ![Python: 3.8+](https://img.shields.io/badge/python-3.8+-green.svg)
 ![FastAPI](https://img.shields.io/badge/framework-FastAPI-009688.svg)
-![Version: 1.1.0](https://img.shields.io/badge/version-1.1.0-brightgreen.svg)
+![Version: 1.2.0](https://img.shields.io/badge/version-1.2.0-brightgreen.svg)
 
-为 Z.AI 提供 OpenAI 和 Anthropic API 兼容接口的轻量级代理服务，支持 GLM-4.5 系列模型的完整功能。
+轻量级 OpenAI API 兼容代理服务，通过 Claude Code Router 接入 Z.AI，支持 GLM-4.5 系列模型的完整功能。
 
 ## ✨ 核心特性
 
 - 🔌 **完全兼容 OpenAI API** - 无缝集成现有应用
-- 🎭 **兼容 Anthropic API** - 支持 Claude CLI 客户端直接接入
+- 🤖 **Claude Code 支持** - 通过 Claude Code Router 工具接入 Claude Code
 - 🚀 **高性能流式响应** - Server-Sent Events (SSE) 支持
-- 🛠️ **Function Call 支持** - 完整的工具调用功能
+- 🛠️ **增强工具调用** - 改进的 Function Call 实现
 - 🧠 **思考模式支持** - 智能处理模型推理过程
 - 🔍 **搜索模型集成** - GLM-4.5-Search 网络搜索能力
 - 🐳 **Docker 部署** - 一键容器化部署
 - 🛡️ **会话隔离** - 匿名模式保护隐私
-- 🔧 **高度可配置** - 环境变量灵活配置
-- 📊 **多模型架构** - 灵活的上游模型映射机制
+- 🔧 **灵活配置** - 环境变量灵活配置
+- 📊 **多模型映射** - 智能上游模型路由
 
 ## 🚀 快速开始
 
@@ -69,28 +69,6 @@ response = client.chat.completions.create(
 print(response.choices[0].message.content)
 ```
 
-#### Anthropic API 客户端
-
-```python
-import anthropic
-
-# 初始化客户端
-client = anthropic.Anthropic(
-    base_url="http://localhost:8080/v1",
-    api_key="your-anthropic-token"  # 替换为你的 ANTHROPIC_API_KEY
-)
-
-# 普通对话
-message = client.messages.create(
-    model="GLM-4.5",
-    max_tokens=1024,
-    messages=[
-        {"role": "user", "content": "你好，介绍一下 Python"}
-    ]
-)
-
-print(message.content[0].text)
-```
 
 ### Docker 部署
 
@@ -163,8 +141,7 @@ for chunk in response:
 
 | 变量名 | 默认值 | 说明 |
 |--------|--------|------|
-| `AUTH_TOKEN` | `sk-your-api-key` | 客户端认证密钥（OpenAI 和 Anthropic 共用） |
-| `ANTHROPIC_API_KEY` | `sk-your-api-key` | Anthropic API 认证密钥（默认使用 AUTH_TOKEN） |
+| `AUTH_TOKEN` | `sk-your-api-key` | 客户端认证密钥 |
 | `API_ENDPOINT` | `https://chat.z.ai/api/chat/completions` | 上游 API 地址 |
 | `LISTEN_PORT` | `8080` | 服务监听端口 |
 | `PRIMARY_MODEL` | `GLM-4.5` | 主要模型名称 |
@@ -244,8 +221,66 @@ if response.choices[0].message.tool_calls:
 **Q: 如何获取 AUTH_TOKEN？**
 A: `AUTH_TOKEN` 为自己自定义的api key，在环境变量中配置，需要保证客户端与服务端一致。
 
-**Q: ANTHROPIC_API_KEY 如何配置？**
-A: 默认使用 `AUTH_TOKEN` 的值，两个 API 使用相同的认证密钥。如需分开配置，可单独设置 `ANTHROPIC_API_KEY` 环境变量。
+**Q: 如何通过 Claude Code 使用本服务？**
+
+A: 复制 [zai.js 文件](https://gist.githubusercontent.com/musistudio/b35402d6f9c95c64269c7666b8405348/raw/f108d66fa050f308387938f149a2b14a295d29e9/gistfile1.txt) 放在`.claude-code-router\\plugins`目录下，配置 Claude Code Router 指向本服务地址，使用 `AUTH_TOKEN` 进行认证。
+
+示例配置：
+```json
+{
+  "LOG": false,
+  "LOG_LEVEL": "debug",
+  "CLAUDE_PATH": "",
+  "HOST": "127.0.0.1",
+  "PORT": 3456,
+  "APIKEY": "",
+  "API_TIMEOUT_MS": "600000",
+  "PROXY_URL": "",
+  "transformers": [
+    {
+      "name": "zai",
+      "path": "C:\\Users\\Administrator\\.claude-code-router\\plugins\\zai.js",
+      "options": {}
+    }
+  ],
+  "Providers": [
+    {
+      "name": "GLM",
+      "api_base_url": "http://127.0.0.1:8080/v1/chat/completions",
+      "api_key": "sk-your-api-key",
+      "models": [
+        "GLM-4.5",
+        "GLM-4.5-Air"
+      ],
+      "transformers": {
+        "use": [
+          "zai"
+        ]
+      }
+    }
+  ],
+  "StatusLine": {
+    "enabled": false,
+    "currentStyle": "default",
+    "default": {
+      "modules": []
+    },
+    "powerline": {
+      "modules": []
+    }
+  },
+  "Router": {
+    "default": "GLM,GLM-4.5",
+    "background": "GLM,GLM-4.5",
+    "think": "GLM,GLM-4.5",
+    "longContext": "GLM,GLM-4.5",
+    "longContextThreshold": 60000,
+    "webSearch": "GLM,GLM-4.5",
+    "image": "GLM,GLM-4.5"
+  },
+  "CUSTOM_ROUTER_PATH": ""
+}
+```
 
 **Q: 匿名模式是什么？**
 A: 匿名模式使用临时 token，避免对话历史共享，保护隐私。
@@ -256,8 +291,8 @@ A: 通过智能提示注入实现，将工具定义转换为系统提示。
 **Q: 支持哪些 OpenAI 功能？**
 A: 支持聊天完成、模型列表、流式响应、工具调用等核心功能。
 
-**Q: 支持 Anthropic API 的哪些功能？**
-A: 支持 messages 创建、流式响应、系统提示等核心功能。
+**Q: Function Call 如何优化？**
+A: 改进了工具调用的请求响应结构，支持更复杂的工具链调用和并行执行。
 
 **Q: 如何选择合适的模型？**
 A: 
@@ -274,34 +309,36 @@ A: 通过环境变量配置，推荐使用 `.env` 文件。
 ```
 ┌──────────────┐      ┌─────────────────────────┐      ┌─────────────────┐
 │   OpenAI     │      │                         │      │                 │
-│  Client      │────▶│    FastAPI Router       │────▶│   Z.AI API      │
+│  Client      │────▶│    FastAPI Server       │────▶│   Z.AI API      │
 └──────────────┘      │                         │      │                 │
 ┌──────────────┐      │ ┌─────────────────────┐ │      │ ┌─────────────┐ │
-│  Anthropic   │────▶│ │   OpenAI Endpoint   │ │      │ │0727-360B-API│ │
-│  Client      │      │ └─────────────────────┘ │      │ └─────────────┘ │
+│ Claude Code  │      │ │ /v1/chat/completions│ │      │ │0727-360B-API│ │
+│   Router     │────▶│ └─────────────────────┘ │      │ └─────────────┘ │
 └──────────────┘      │ ┌─────────────────────┐ │      │ ┌─────────────┐ │
-                      │ │  Anthropic Endpoint │ │────▶│ │0727-106B-API│ │
+                      │ │    /v1/models       │ │────▶│ │0727-106B-API│ │
                       │ └─────────────────────┘ │      │ └─────────────┘ │
                       │ ┌─────────────────────┐ │      │                 │
-                      │ │   Models Endpoint   │ │      └─────────────────┘
+                      │ │  Enhanced Tools     │ │      └─────────────────┘
                       │ └─────────────────────┘ │
                       └─────────────────────────┘
-                              Proxy Server
+                           OpenAI Compatible API
 ```
 
 ### 核心组件
 
 - **FastAPI** - 高性能 Web 框架，支持异步处理
-- **Pydantic** - 数据验证和序列化，确保 API 兼容性
+- **Pydantic** - 数据验证和序列化，确保 API 兼容性  
 - **Uvicorn** - ASGI 服务器，提供高性能服务
-- **Requests** - HTTP 客户端，与上游 API 通信
+- **httpx** - 现代 HTTP 客户端，支持异步请求
+- **SSE Parser** - 流式响应处理，优化实时交互
 
 ### 架构特点
 
 - **模块化设计** - 清晰的目录结构，易于维护和扩展
-- **多协议支持** - 同时支持 OpenAI 和 Anthropic API 协议
-- **动态路由** - 根据请求模型自动选择上游服务
-- **流式处理** - 完整支持 SSE 流式响应
+- **标准 OpenAI 协议** - 完全兼容 OpenAI API v1 规范
+- **智能模型路由** - 根据模型特性自动选择最优上游
+- **增强工具调用** - 改进的 Function Call 处理机制
+- **流式处理** - 优化的 SSE 流式响应实现
 - **类型安全** - 基于 Pydantic 的严格类型检查
 
 ### 项目结构
@@ -309,27 +346,27 @@ A: 通过环境变量配置，推荐使用 `.env` 文件。
 ```
 z.ai2api_python/
 ├── app/
-│   ├── api/
-│   │   ├── __init__.py
-│   │   ├── openai.py          # OpenAI API 路由
-│   │   └── anthropic.py       # Anthropic API 路由
 │   ├── core/
 │   │   ├── __init__.py
 │   │   ├── config.py          # 配置管理
+│   │   ├── openai.py          # OpenAI API 实现
 │   │   └── response_handlers.py  # 响应处理器
 │   ├── models/
 │   │   ├── __init__.py
-│   │   └── schemas.py         # 数据模型定义
+│   │   └── schemas.py         # Pydantic 模型定义
 │   ├── utils/
 │   │   ├── __init__.py
-│   │   ├── helpers.py         # 工具函数
-│   │   ├── tools.py           # Function Call 处理
-│   │   └── sse_parser.py      # SSE 解析器
+│   │   ├── helpers.py         # 辅助函数
+│   │   ├── tools.py           # 增强工具调用处理
+│   │   └── sse_parser.py      # SSE 流式解析器
 │   └── __init__.py
-├── tests/                     # 测试文件
-├── deploy/                    # 部署配置
-├── main.py                    # 应用入口
-├── requirements.txt           # 依赖列表
+├── tests/                     # 单元测试
+│   ├── test_tool_call.py      # 工具调用测试
+│   └── test_function_call.py  # Function Call 测试
+├── deploy/                    # Docker 部署配置
+├── main.py                    # FastAPI 应用入口
+├── requirements.txt           # Python 依赖
+├── .env.example              # 环境变量示例
 └── README.md                  # 项目文档
 ```
 

app/__init__.py

@@ -2,6 +2,6 @@
 Application package initialization
 """
 
-from app import api, core, models, utils
+from app import core, models, utils
 
-__all__ = ["api", "core", "models", "utils"]
+__all__ = ["core", "models", "utils"]

app/api/__init__.py

@@ -1,7 +0,0 @@
-"""
-API module initialization
-"""
-
-from app.api import openai, anthropic
-
-__all__ = ["openai", "anthropic"]

app/api/anthropic.py

@@ -1,276 +0,0 @@
-"""
-Anthropic API compatibility endpoints
-"""
-
-import json
-import time
-import uuid
-from typing import Generator
-import requests
-from fastapi import APIRouter, Header, HTTPException
-from fastapi.responses import StreamingResponse
-
-from app.core.config import settings
-from app.models.schemas import (
-    AnthropicRequest, Message, UpstreamRequest, ModelItem,
-    ContentBlock
-)
-from app.utils.helpers import debug_log, generate_request_ids, get_auth_token, get_browser_headers, transform_thinking_content
-
-router = APIRouter()
-
-
-def stream_anthropic_generator(upstream_response: requests.Response, request_id: str, requested_model: str) -> Generator[str, None, None]:
-    """生成 Anthropic 兼容的流式响应事件"""
-    usage = {"input_tokens": 0, "output_tokens": 0}
-    
-    start_event = {
-        "type": "message_start",
-        "message": {
-            "id": request_id,
-            "type": "message",
-            "role": "assistant",
-            "content": [],
-            "model": requested_model,
-            "stop_reason": None,
-            "stop_sequence": None,
-            "usage": usage
-        }
-    }
-    yield f"event: {start_event['type']}\ndata: {json.dumps(start_event['message'])}\n\n"
-    
-    # 发送 content_block_start 事件
-    content_start_data = {
-        "type": "content_block_start",
-        "index": 0,
-        "content_block": {
-            "type": "text",
-            "text": ""
-        }
-    }
-    yield f"event: content_block_start\ndata: {json.dumps(content_start_data)}\n\n"
-    
-    # 处理上游响应
-    for line in upstream_response.iter_lines():
-        if not line.startswith(b"data:"): continue
-        data_str = line[5:].strip()
-        if not data_str: continue
-        try:
-            data = json.loads(data_str.decode('utf-8'))
-            delta_content = data.get("data", {}).get("delta_content", "")
-            phase = data.get("data", {}).get("phase", "")
-            
-            # 处理内容增量
-            if delta_content:
-                out_content = transform_thinking_content(delta_content) if phase == "thinking" else delta_content
-                if out_content:
-                    usage["output_tokens"] += len(out_content) // 4  # 简单估算
-                    delta_data = {
-                        "type": "content_block_delta",
-                        "index": 0,
-                        "delta": {
-                            "type": "text_delta",
-                            "text": out_content
-                        }
-                    }
-                    yield f"event: content_block_delta\ndata: {json.dumps(delta_data)}\n\n"
-            
-            # 处理结束
-            if data.get("data", {}).get("done", False) or phase == "done":
-                # 发送 content_block_stop
-                content_stop_data = {
-                    "type": "content_block_stop",
-                    "index": 0
-                }
-                yield f"event: content_block_stop\ndata: {json.dumps(content_stop_data)}\n\n"
-                
-                # 发送 message_delta
-                message_delta_data = {
-                    "type": "message_delta",
-                    "delta": {
-                        "stop_reason": "end_turn",
-                        "stop_sequence": None,
-                        "usage": {
-                            "input_tokens": usage["input_tokens"],
-                            "output_tokens": usage["output_tokens"]
-                        }
-                    }
-                }
-                yield f"event: message_delta\ndata: {json.dumps(message_delta_data)}\n\n"
-                
-                # 发送 message_stop
-                yield f"event: message_stop\ndata: {json.dumps({'type': 'message_stop'})}\n\n"
-                break
-                
-        except json.JSONDecodeError:
-            continue
-
-
-@router.post("/v1/messages")
-async def handle_anthropic_message(
-    req: AnthropicRequest,
-    x_api_key: str = Header(None, alias="x-api-key"),
-    authorization: str = Header(None, alias="authorization")
-):
-    """Handle Anthropic message requests"""
-    debug_log("收到 Anthropic message 请求")
-    
-    # 验证 API key (skip if SKIP_AUTH_TOKEN is enabled)
-    if not settings.SKIP_AUTH_TOKEN:
-        api_key = None
-        if x_api_key:
-            api_key = x_api_key
-        elif authorization and authorization.startswith("Bearer "):
-            api_key = authorization[7:]
-        
-        if not api_key or api_key != settings.ANTHROPIC_API_KEY:
-            debug_log(f"无效的 API key: {api_key}")
-            raise HTTPException(status_code=401, detail="Invalid API key")
-        
-        debug_log(f"API key 验证通过")
-    else:
-        debug_log("SKIP_AUTH_TOKEN已启用，跳过API key验证")
-    debug_log(f"请求解析成功 - 模型: {req.model}, 流式: {req.stream}, 消息数: {len(req.messages)}")
-    
-    # 确定上游模型和功能
-    upstream_model = "GLM-4.5"
-    if req.model == settings.THINKING_MODEL:
-        upstream_model = "GLM-4.5-Thinking"
-    elif req.model == settings.SEARCH_MODEL:
-        upstream_model = "GLM-4.5-Search"
-    
-    debug_log(f"收到请求 (模型: {req.model}) -> 代理到上游 (模型: {upstream_model})")
-    
-    # 生成 ID
-    chat_id, msg_id = generate_request_ids()
-    
-    # 转换消息格式
-    openai_messages = []
-    if req.system:
-        # 处理两种格式的 system 内容
-        if isinstance(req.system, str):
-            # 字符串格式
-            system_content = req.system
-        else:
-            # 对象数组格式
-            system_content = ""
-            for block in req.system:
-                if block.type == "text":
-                    system_content += block.text
-        
-        openai_messages.append({"role": "system", "content": system_content})
-    
-    for msg in req.messages:
-        # 处理两种格式的内容
-        if isinstance(msg.content, str):
-            # 字符串格式
-            text_content = msg.content
-        else:
-            # 对象数组格式
-            text_content = ""
-            for block in msg.content:
-                if block.type == "text":
-                    text_content += block.text
-        
-        openai_messages.append({
-            "role": msg.role,
-            "content": text_content
-        })
-    
-    # 构建上游请求
-    upstream_messages = []
-    for msg in openai_messages:
-        content = msg.get("content", "")
-        if content is None:
-            content = ""
-        upstream_messages.append(Message(
-            role=msg["role"],
-            content=content
-        ))
-    
-    upstream_req = UpstreamRequest(
-        stream=True,  # 总是使用上游的流式
-        chat_id=chat_id,
-        id=msg_id,
-        model="0727-360B-API",  # 实际的上游模型 ID
-        messages=upstream_messages,
-        params={},
-        features={"enable_thinking": True},
-        background_tasks={
-            "title_generation": False,
-            "tags_generation": False,
-        },
-        mcp_servers=[],
-        model_item=ModelItem(
-            id="0727-360B-API",
-            name="GLM-4.5",
-            owned_by="openai"
-        ),
-        tool_servers=[],
-        variables={
-            "{{USER_NAME}}": "User",
-            "{{USER_LOCATION}}": "Unknown",
-            "{{CURRENT_DATETIME}}": time.strftime("%Y-%m-%d %H:%M:%S"),
-        }
-    )
-    
-    # 获取认证 token
-    auth_token = get_auth_token()
-    
-    try:
-        # 调用上游 API
-        headers = get_browser_headers(chat_id)
-        headers["Authorization"] = f"Bearer {auth_token}"
-        
-        response = requests.post(
-            settings.API_ENDPOINT,
-            json=upstream_req.model_dump(exclude_none=True),
-            headers=headers,
-            timeout=60.0,
-            stream=True
-        )
-        response.raise_for_status()
-    except requests.HTTPError as e:
-        debug_log(f"上游 API 返回错误状态: {e.response.status_code}, 响应: {e.response.text}")
-        raise HTTPException(status_code=502, detail="Upstream API error")
-    except requests.RequestException as e:
-        debug_log(f"请求上游 API 失败: {e}")
-        raise HTTPException(status_code=502, detail=f"Failed to call upstream API: {e}")
-    
-    request_id = f"msg_{uuid.uuid4().hex}"
-    
-    if req.stream:
-        # 流式响应
-        return StreamingResponse(
-            stream_anthropic_generator(response, request_id, req.model),
-            media_type="text/event-stream",
-            headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}
-        )
-    else:
-        # 非流式响应
-        full_content = ""
-        for line in response.iter_lines():
-            if not line.startswith(b"data:"): continue
-            data_str = line[5:].strip()
-            if not data_str: continue
-            try:
-                data = json.loads(data_str.decode('utf-8'))
-                delta_content = data.get("data", {}).get("delta_content", "")
-                phase = data.get("data", {}).get("phase", "")
-                if delta_content:
-                    out_content = transform_thinking_content(delta_content) if phase == "thinking" else delta_content
-                    if out_content: full_content += out_content
-                if data.get("data", {}).get("done", False) or phase == "done":
-                    break
-            except json.JSONDecodeError:
-                continue
-        
-        return {
-            "id": request_id,
-            "type": "message",
-            "role": "assistant",
-            "model": req.model,
-            "content": [{"type": "text", "text": full_content}],
-            "stop_reason": "end_turn",
-            "usage": {"input_tokens": 0, "output_tokens": len(full_content) // 4}
-        }

app/core/__init__.py

@@ -2,6 +2,6 @@
 Core module initialization
 """
 
-from app.core import config, response_handlers
+from app.core import config, response_handlers, openai
 
-__all__ = ["config", "response_handlers"]
+__all__ = ["config", "response_handlers", "openai"]

app/core/config.py

@@ -13,7 +13,6 @@ class Settings(BaseSettings):
     # API Configuration
     API_ENDPOINT: str = os.getenv("API_ENDPOINT", "https://chat.z.ai/api/chat/completions")
     AUTH_TOKEN: str = os.getenv("AUTH_TOKEN", "sk-your-api-key")
-    ANTHROPIC_API_KEY: str = os.getenv("ANTHROPIC_API_KEY", AUTH_TOKEN)
     BACKUP_TOKEN: str = os.getenv("BACKUP_TOKEN", "eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjMxNmJjYjQ4LWZmMmYtNGExNS04NTNkLWYyYTI5YjY3ZmYwZiIsImVtYWlsIjoiR3Vlc3QtMTc1NTg0ODU4ODc4OEBndWVzdC5jb20ifQ.PktllDySS3trlyuFpTeIZf-7hl8Qu1qYF3BxjgIul0BrNux2nX9hVzIjthLXKMWAf9V0qM8Vm_iyDqkjPGsaiQ")
     
     # Model Configuration

app/{api → core}/openai.py

@@ -14,7 +14,7 @@ from app.models.schemas import (
     ModelsResponse, Model
 )
 from app.utils.helpers import debug_log, generate_request_ids, get_auth_token
-from app.utils.tools import process_messages_with_tools
+from app.utils.tools import process_messages_with_tools, content_to_string
 from app.core.response_handlers import StreamResponseHandler, NonStreamResponseHandler
 
 router = APIRouter()
@@ -89,10 +89,7 @@ async def chat_completions(
         # Convert back to Message objects
         upstream_messages: List[Message] = []
         for msg in processed_messages:
-            content = msg.get("content")
-            # Ensure content is not None for Message model
-            if content is None:
-                content = ""
+            content = content_to_string(msg.get("content"))
             
             upstream_messages.append(Message(
                 role=msg["role"],

app/core/response_handlers.py

@@ -205,26 +205,28 @@ class StreamResponseHandler(ResponseHandler):
     
     def _send_end_chunk(self) -> Generator[str, None, None]:
         """Send end chunk and DONE signal"""
+        finish_reason = "stop"
+        
         if self.has_tools:
             # Try to extract tool calls from buffered content
             self.tool_calls = extract_tool_invocations(self.buffered_content)
             
             if self.tool_calls:
-                # Send tool calls
-                tool_calls_list = []
+                # Send tool calls with proper format
                 for i, tc in enumerate(self.tool_calls):
-                    tool_calls_list.append({
+                    tool_call_delta = {
                         "index": i,
                         "id": tc.get("id"),
                         "type": tc.get("type", "function"),
                         "function": tc.get("function", {}),
-                    })
+                    }
+                    
+                    out_chunk = create_openai_response_chunk(
+                        model=settings.PRIMARY_MODEL,
+                        delta=Delta(tool_calls=[tool_call_delta])
+                    )
+                    yield f"data: {out_chunk.model_dump_json()}\n\n"
                 
-                out_chunk = create_openai_response_chunk(
-                    model=settings.PRIMARY_MODEL,
-                    delta=Delta(tool_calls=tool_calls_list)
-                )
-                yield f"data: {out_chunk.model_dump_json()}\n\n"
                 finish_reason = "tool_calls"
             else:
                 # Send regular content
@@ -235,9 +237,6 @@ class StreamResponseHandler(ResponseHandler):
                         delta=Delta(content=trimmed_content)
                     )
                     yield f"data: {content_chunk.model_dump_json()}\n\n"
-                finish_reason = "stop"
-        else:
-            finish_reason = "stop"
         
         # Send final chunk
         end_chunk = create_openai_response_chunk(
@@ -305,9 +304,12 @@ class NonStreamResponseHandler(ResponseHandler):
                 # Content must be null when tool_calls are present (OpenAI spec)
                 message_content = None
                 finish_reason = "tool_calls"
+                debug_log(f"提取到工具调用: {json.dumps(tool_calls, ensure_ascii=False)}")
             else:
                 # Remove tool JSON from content
                 message_content = remove_tool_json_content(final_content)
+                if not message_content:
+                    message_content = final_content  # 保留原内容如果清理后为空
         
         # Build response
         response_data = OpenAIResponse(

app/models/schemas.py

@@ -6,10 +6,16 @@ from typing import Dict, List, Optional, Any, Union, Literal
 from pydantic import BaseModel
 
 
+class ContentPart(BaseModel):
+    """Content part model for OpenAI's new content format"""
+    type: str
+    text: Optional[str] = None
+
+
 class Message(BaseModel):
     """Chat message model"""
     role: str
-    content: Optional[str] = None
+    content: Optional[Union[str, List[ContentPart]]] = None
     reasoning_content: Optional[str] = None
     tool_calls: Optional[List[Dict[str, Any]]] = None
 
@@ -125,21 +131,3 @@ class ModelsResponse(BaseModel):
     data: List[Model]
 
 
-# Anthropic API Models
-class ContentBlock(BaseModel):
-    type: str
-    text: str
-
-
-class AnthropicMessage(BaseModel):
-    role: Literal["user", "assistant"]
-    content: Union[str, List[ContentBlock]]
-
-
-class AnthropicRequest(BaseModel):
-    model: str
-    messages: List[AnthropicMessage]
-    system: Optional[Union[str, List[ContentBlock]]] = None
-    max_tokens: int = 1024
-    stream: bool = False
-    temperature: Optional[float] = None

app/utils/sse_parser.py

@@ -6,16 +6,13 @@ import json
 from typing import Dict, Any, Generator, Optional, Type
 import requests
 
-from app.core.config import settings
-from app.models.schemas import UpstreamData
-
 
 class SSEParser:
     """Server-Sent Events parser for streaming responses"""
-    
+
     def __init__(self, response: requests.Response, debug_mode: bool = False):
         """Initialize SSE parser
-        
+
         Args:
             response: requests.Response object with stream=True
             debug_mode: Enable debug logging
@@ -24,7 +21,7 @@ class SSEParser:
         self.debug_mode = debug_mode
         self.buffer = ""
         self.line_count = 0
-    
+
     def debug_log(self, format_str: str, *args) -> None:
         """Log debug message if debug mode is enabled"""
         if self.debug_mode:
@@ -32,112 +29,99 @@ class SSEParser:
                 print(f"[SSE_PARSER] {format_str % args}")
             else:
                 print(f"[SSE_PARSER] {format_str}")
-    
+
     def iter_events(self) -> Generator[Dict[str, Any], None, None]:
         """Iterate over SSE events
-        
+
         Yields:
             dict: Parsed SSE event data
         """
         self.debug_log("开始解析 SSE 流")
-        
+
         for line in self.response.iter_lines():
             self.line_count += 1
-            
+
             # Skip empty lines
             if not line:
                 continue
-            
+
             # Decode bytes
             if isinstance(line, bytes):
                 try:
-                    line = line.decode('utf-8')
+                    line = line.decode("utf-8")
                 except UnicodeDecodeError:
                     self.debug_log(f"第{self.line_count}行解码失败，跳过")
                     continue
-            
+
             # Skip comment lines
-            if line.startswith(':'):
+            if line.startswith(":"):
                 continue
-            
+
             # Parse field-value pairs
-            if ':' in line:
-                field, value = line.split(':', 1)
+            if ":" in line:
+                field, value = line.split(":", 1)
                 field = field.strip()
                 value = value.lstrip()
-                
-                if field == 'data':
+
+                if field == "data":
                     self.debug_log(f"收到数据 (第{self.line_count}行): {value}")
-                    
+
                     # Try to parse JSON
                     try:
                         data = json.loads(value)
-                        yield {
-                            'type': 'data',
-                            'data': data,
-                            'raw': value
-                        }
+                        yield {"type": "data", "data": data, "raw": value}
                     except json.JSONDecodeError:
-                        yield {
-                            'type': 'data',
-                            'data': value,
-                            'raw': value,
-                            'is_json': False
-                        }
-                
-                elif field == 'event':
-                    yield {'type': 'event', 'event': value}
-                
-                elif field == 'id':
-                    yield {'type': 'id', 'id': value}
-                
-                elif field == 'retry':
+                        yield {"type": "data", "data": value, "raw": value, "is_json": False}
+
+                elif field == "event":
+                    yield {"type": "event", "event": value}
+
+                elif field == "id":
+                    yield {"type": "id", "id": value}
+
+                elif field == "retry":
                     try:
                         retry = int(value)
-                        yield {'type': 'retry', 'retry': retry}
+                        yield {"type": "retry", "retry": retry}
                     except ValueError:
                         self.debug_log(f"无效的 retry 值: {value}")
-    
+
     def iter_data_only(self) -> Generator[Dict[str, Any], None, None]:
         """Iterate only over data events"""
         for event in self.iter_events():
-            if event['type'] == 'data':
+            if event["type"] == "data":
                 yield event
-    
+
     def iter_json_data(self, model_class: Optional[Type] = None) -> Generator[Dict[str, Any], None, None]:
         """Iterate only over JSON data events with optional validation
-        
+
         Args:
             model_class: Optional Pydantic model class for validation
-            
+
         Yields:
             dict: JSON data events
         """
         for event in self.iter_events():
-            if event['type'] == 'data' and event.get('is_json', True):
+            if event["type"] == "data" and event.get("is_json", True):
                 try:
                     if model_class:
-                        data = model_class.model_validate_json(event['raw'])
-                        yield {
-                            'type': 'data',
-                            'data': data,
-                            'raw': event['raw']
-                        }
+                        data = model_class.model_validate_json(event["raw"])
+                        yield {"type": "data", "data": data, "raw": event["raw"]}
                     else:
                         yield event
                 except Exception as e:
                     self.debug_log(f"数据验证失败: {e}")
                     continue
-    
+
     def close(self) -> None:
         """Close the response connection"""
-        if hasattr(self.response, 'close'):
+        if hasattr(self.response, "close"):
             self.response.close()
-    
+
     def __enter__(self):
         """Context manager entry"""
         return self
-    
+
     def __exit__(self, exc_type, exc_val, exc_tb) -> None:
         """Context manager exit"""
-        self.close()
+        self.close()

app/utils/tools.py

@@ -10,28 +10,43 @@ from typing import Dict, List, Optional, Any
 from app.core.config import settings
 
 
+def content_to_string(content: Any) -> str:
+    """Convert content from various formats to string (following app.py pattern)"""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts = []
+        for p in content:
+            if isinstance(p, dict) and p.get("type") == "text":
+                parts.append(p.get("text", ""))
+            elif isinstance(p, str):
+                parts.append(p)
+        return " ".join(parts)
+    return ""
+
+
 def generate_tool_prompt(tools: List[Dict[str, Any]]) -> str:
     """Generate tool injection prompt with enhanced formatting"""
     if not tools:
         return ""
-    
+
     tool_definitions = []
     for tool in tools:
         if tool.get("type") != "function":
             continue
-            
+
         function_spec = tool.get("function", {}) or {}
         function_name = function_spec.get("name", "unknown")
         function_description = function_spec.get("description", "")
         parameters = function_spec.get("parameters", {}) or {}
-        
+
         # Create structured tool definition
         tool_info = [f"## {function_name}", f"**Purpose**: {function_description}"]
-        
+
         # Add parameter details
         parameter_properties = parameters.get("properties", {}) or {}
         required_parameters = set(parameters.get("required", []) or [])
-        
+
         if parameter_properties:
             tool_info.append("**Parameters**:")
             for param_name, param_details in parameter_properties.items():
@@ -39,111 +54,103 @@ def generate_tool_prompt(tools: List[Dict[str, Any]]) -> str:
                 param_desc = (param_details or {}).get("description", "")
                 requirement_flag = "**Required**" if param_name in required_parameters else "*Optional*"
                 tool_info.append(f"- `{param_name}` ({param_type}) - {requirement_flag}: {param_desc}")
-        
+
         tool_definitions.append("\n".join(tool_info))
-    
+
     if not tool_definitions:
         return ""
-    
+
     # Build comprehensive tool prompt
     prompt_template = (
-        "\n\n# AVAILABLE FUNCTIONS\n" +
-        "\n\n---\n".join(tool_definitions) +
-        "\n\n# USAGE INSTRUCTIONS\n"
+        "\n\n# AVAILABLE FUNCTIONS\n" + "\n\n---\n".join(tool_definitions) + "\n\n# USAGE INSTRUCTIONS\n"
         "When you need to execute a function, respond ONLY with a JSON object containing tool_calls:\n"
         "```json\n"
         "{\n"
         '  "tool_calls": [\n'
         "    {\n"
-        '      "id": "call_" + unique_id,\n'
+        '      "id": "call_xxx",\n'
         '      "type": "function",\n'
         '      "function": {\n'
         '        "name": "function_name",\n'
-        '        "arguments": {\n'
-        '          "param1": "value1"\n'
-        '        }\n'
+        '        "arguments": "{\\"param1\\": \\"value1\\"}"\n'
         "      }\n"
         "    }\n"
         "  ]\n"
         "}\n"
         "```\n"
-        "Important: No explanatory text before or after the JSON.\n"
+        "Important: No explanatory text before or after the JSON. The 'arguments' field must be a JSON string, not an object.\n"
     )
-    
+
     return prompt_template
 
 
 def process_messages_with_tools(
-    messages: List[Dict[str, Any]], 
-    tools: Optional[List[Dict[str, Any]]] = None,
-    tool_choice: Optional[Any] = None
+    messages: List[Dict[str, Any]], tools: Optional[List[Dict[str, Any]]] = None, tool_choice: Optional[Any] = None
 ) -> List[Dict[str, Any]]:
     """Process messages and inject tool prompts"""
     processed: List[Dict[str, Any]] = []
-    
+
     if tools and settings.TOOL_SUPPORT and (tool_choice != "none"):
         tools_prompt = generate_tool_prompt(tools)
         has_system = any(m.get("role") == "system" for m in messages)
-        
+
         if has_system:
             for m in messages:
                 if m.get("role") == "system":
                     mm = dict(m)
-                    content = mm.get("content", "")
-                    if content is None:
-                        content = ""
+                    content = content_to_string(mm.get("content", ""))
                     mm["content"] = content + tools_prompt
                     processed.append(mm)
                 else:
                     processed.append(m)
         else:
             processed = [{"role": "system", "content": "你是一个有用的助手。" + tools_prompt}] + messages
-        
+
         # Add tool choice hints
         if tool_choice in ("required", "auto"):
             if processed and processed[-1].get("role") == "user":
                 last = dict(processed[-1])
-                content = last.get("content", "")
-                if content is None:
-                    content = ""
+                content = content_to_string(last.get("content", ""))
                 last["content"] = content + "\n\n请根据需要使用提供的工具函数。"
                 processed[-1] = last
         elif isinstance(tool_choice, dict) and tool_choice.get("type") == "function":
             fname = (tool_choice.get("function") or {}).get("name")
             if fname and processed and processed[-1].get("role") == "user":
                 last = dict(processed[-1])
-                content = last.get("content", "")
-                if content is None:
-                    content = ""
+                content = content_to_string(last.get("content", ""))
                 last["content"] = content + f"\n\n请使用 {fname} 函数来处理这个请求。"
                 processed[-1] = last
     else:
         processed = list(messages)
-    
+
     # Handle tool/function messages
     final_msgs: List[Dict[str, Any]] = []
     for m in processed:
         role = m.get("role")
         if role in ("tool", "function"):
             tool_name = m.get("name", "unknown")
-            tool_content = m.get("content", "")
+            tool_content = content_to_string(m.get("content", ""))
             if isinstance(tool_content, dict):
                 tool_content = json.dumps(tool_content, ensure_ascii=False)
-            elif tool_content is None:
-                tool_content = ""
-            
+
             # 确保内容不为空且不包含 None
             content = f"工具 {tool_name} 返回结果:\n```json\n{tool_content}\n```"
             if not content.strip():
                 content = f"工具 {tool_name} 执行完成"
-                
-            final_msgs.append({
-                "role": "assistant",
-                "content": content,
-            })
+
+            final_msgs.append(
+                {
+                    "role": "assistant",
+                    "content": content,
+                }
+            )
         else:
-            final_msgs.append(m)
-    
+            # For regular messages, ensure content is string format
+            final_msg = dict(m)
+            content = content_to_string(final_msg.get("content", ""))
+            final_msg["content"] = content
+            final_msgs.append(final_msg)
+
     return final_msgs
 
 
@@ -157,10 +164,10 @@ def extract_tool_invocations(text: str) -> Optional[List[Dict[str, Any]]]:
     """Extract tool invocations from response text"""
     if not text:
         return None
-    
+
     # Limit scan size for performance
-    scannable_text = text[:settings.SCAN_LIMIT]
-    
+    scannable_text = text[: settings.SCAN_LIMIT]
+
     # Attempt 1: Extract from JSON code blocks
     json_blocks = TOOL_CALL_FENCE_PATTERN.findall(scannable_text)
     for json_block in json_blocks:
@@ -168,10 +175,20 @@ def extract_tool_invocations(text: str) -> Optional[List[Dict[str, Any]]]:
             parsed_data = json.loads(json_block)
             tool_calls = parsed_data.get("tool_calls")
             if tool_calls and isinstance(tool_calls, list):
+                # Ensure arguments field is a string
+                for tc in tool_calls:
+                    if "function" in tc:
+                        func = tc["function"]
+                        if "arguments" in func:
+                            if isinstance(func["arguments"], dict):
+                                # Convert dict to JSON string
+                                func["arguments"] = json.dumps(func["arguments"], ensure_ascii=False)
+                            elif not isinstance(func["arguments"], str):
+                                func["arguments"] = json.dumps(func["arguments"], ensure_ascii=False)
                 return tool_calls
         except (json.JSONDecodeError, AttributeError):
             continue
-    
+
     # Attempt 2: Extract inline JSON objects
     inline_match = TOOL_CALL_INLINE_PATTERN.search(scannable_text)
     if inline_match:
@@ -180,10 +197,20 @@ def extract_tool_invocations(text: str) -> Optional[List[Dict[str, Any]]]:
             parsed_data = json.loads(inline_json)
             tool_calls = parsed_data.get("tool_calls")
             if tool_calls and isinstance(tool_calls, list):
+                # Ensure arguments field is a string
+                for tc in tool_calls:
+                    if "function" in tc:
+                        func = tc["function"]
+                        if "arguments" in func:
+                            if isinstance(func["arguments"], dict):
+                                # Convert dict to JSON string
+                                func["arguments"] = json.dumps(func["arguments"], ensure_ascii=False)
+                            elif not isinstance(func["arguments"], str):
+                                func["arguments"] = json.dumps(func["arguments"], ensure_ascii=False)
                 return tool_calls
         except (json.JSONDecodeError, AttributeError):
             pass
-    
+
     # Attempt 3: Parse natural language function calls
     natural_lang_match = FUNCTION_CALL_PATTERN.search(scannable_text)
     if natural_lang_match:
@@ -192,22 +219,22 @@ def extract_tool_invocations(text: str) -> Optional[List[Dict[str, Any]]]:
         try:
             # Validate JSON format
             json.loads(arguments_str)
-            return [{
-                "id": f"invoke_{int(time.time() * 1000000)}",
-                "type": "function",
-                "function": {
-                    "name": function_name,
-                    "arguments": arguments_str
+            return [
+                {
+                    "id": f"call_{int(time.time() * 1000000)}",
+                    "type": "function",
+                    "function": {"name": function_name, "arguments": arguments_str},
                 }
-            }]
+            ]
         except json.JSONDecodeError:
             return None
-    
+
     return None
 
 
 def remove_tool_json_content(text: str) -> str:
     """Remove tool JSON content from response text"""
+
     def remove_tool_call_block(match: re.Match) -> str:
         json_content = match.group(1)
         try:
@@ -217,9 +244,9 @@ def remove_tool_json_content(text: str) -> str:
         except (json.JSONDecodeError, AttributeError):
             pass
         return match.group(0)
-    
+
     # Remove fenced tool JSON blocks
     cleaned_text = TOOL_CALL_FENCE_PATTERN.sub(remove_tool_call_block, text)
     # Remove inline tool JSON
     cleaned_text = TOOL_CALL_INLINE_PATTERN.sub("", cleaned_text)
-    return cleaned_text.strip()
+    return cleaned_text.strip()

main.py

@@ -6,13 +6,13 @@ from fastapi import FastAPI, Request, Response
 from fastapi.middleware.cors import CORSMiddleware
 
 from app.core.config import settings
-from app.api import openai, anthropic
+from app.core import openai
 
 # Create FastAPI app
 app = FastAPI(
     title="OpenAI Compatible API Server",
     description="An OpenAI-compatible API server for Z.AI chat service",
-    version="1.0.0"
+    version="1.0.0",
 )
 
 # Add CORS middleware
@@ -26,7 +26,6 @@ app.add_middleware(
 
 # Include API routers
 app.include_router(openai.router)
-app.include_router(anthropic.router)
 
 
 @app.options("/")
@@ -43,4 +42,5 @@ async def root():
 
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run("main:app", host="0.0.0.0", port=settings.LISTEN_PORT, reload=True)
+
+    uvicorn.run("main:app", host="0.0.0.0", port=settings.LISTEN_PORT, reload=True)

tests/test_anthropic.py

@@ -1,79 +0,0 @@
-# -*- coding: utf-8 -*-
-
-import json
-import requests
-
-# 服务器配置
-BASE_URL = "http://localhost:8080/v1/messages"
-API_KEY = "sk-your-api-key"
-
-test_data = {
-    "model": "GLM-4.5",
-    "messages": [{"role": "user", "content": "你好，这是一个测试"}],
-    "system": [
-        {
-            "type": "text",
-            "text": "You are Claude Code, Anthropic's official CLI for Claude.",
-            "cache_control": {"type": "ephemeral"},
-        }
-    ],
-    "max_tokens": 1024,
-    "stream": False,
-}
-
-
-def test_non_stream():
-    """测试非流式请求"""
-    print("=== 测试非流式请求 ===")
-
-    try:
-        response = requests.post(BASE_URL, headers={"x-api-key": API_KEY}, json=test_data, timeout=30.0)
-
-        print(f"状态码: {response.status_code}")
-
-        if response.status_code == 200:
-            result = response.json()
-            print("响应成功!")
-            print(f"ID: {result.get('id')}")
-            print(f"模型: {result.get('model')}")
-            if result.get("content"):
-                print(f"内容: {result['content'][0]['text']}")
-        else:
-            print("错误响应:")
-            print(response.text)
-
-    except Exception as e:
-        print(f"请求失败: {e}")
-
-
-def test_stream():
-    """测试流式请求"""
-    print("\n=== 测试流式请求 ===")
-
-    stream_data = test_data.copy()
-    stream_data["stream"] = True
-
-    try:
-        response = requests.post(BASE_URL, headers={"x-api-key": API_KEY}, json=stream_data, stream=True, timeout=30.0)
-
-        print(f"状态码: {response.status_code}")
-
-        if response.status_code == 200:
-            print("流式响应内容:")
-            for line in response.iter_lines():
-                if line:
-                    print(f"  {line.decode('utf-8')}")
-        else:
-            print("错误响应:")
-            print(response.text)
-
-    except Exception as e:
-        print(f"请求失败: {e}")
-
-
-if __name__ == "__main__":
-    try:
-        test_non_stream()
-        test_stream()
-    except KeyboardInterrupt:
-        print("\n测试已取消")

tests/test_system_field.py

@@ -1,68 +0,0 @@
-#!/usr/bin/env python3
-"""
-测试 Anthropic API system 字段数组类型支持
-"""
-import json
-import requests
-
-# 测试数据
-test_cases = [
-    {
-        "name": "字符串类型 system",
-        "data": {
-            "model": "GLM-4.5",
-            "messages": [{"role": "user", "content": "你好"}],
-            "system": "你是一个有帮助的助手",
-            "max_tokens": 100
-        }
-    },
-    {
-        "name": "数组类型 system",
-        "data": {
-            "model": "GLM-4.5",
-            "messages": [{"role": "user", "content": "你好"}],
-            "system": [
-                {
-                    "type": "text",
-                    "text": "你是一个有帮助的助手",
-                    "cache_control": {"type": "ephemeral"}
-                }
-            ],
-            "max_tokens": 100
-        }
-    }
-]
-
-def test_system_field():
-    """测试 system 字段的不同格式"""
-    print("=== 测试 system 字段支持 ===\n")
-    
-    for test_case in test_cases:
-        print(f"测试: {test_case['name']}")
-        
-        try:
-            response = requests.post(
-                "http://localhost:8080/v1/messages",
-                headers={"x-api-key": "sk-your-api-key"},
-                json=test_case["data"],
-                timeout=10
-            )
-            
-            if response.status_code == 200:
-                result = response.json()
-                print("✅ 成功")
-                print(f"   消息ID: {result.get('id')}")
-                print(f"   内容预览: {result['content'][0]['text'][:50]}...")
-            else:
-                print(f"❌ 失败 - 状态码: {response.status_code}")
-                print(f"   错误: {response.text}")
-                
-        except Exception as e:
-            print(f"❌ 异常: {e}")
-        
-        print()
-
-if __name__ == "__main__":
-    print("请确保服务器正在运行在 http://localhost:8080")
-    input("按 Enter 开始测试...")
-    test_system_field()

tests/test_tool_call.py

@@ -0,0 +1,145 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+测试工具调用功能
+"""
+
+import json
+import requests
+
+# 配置
+BASE_URL = "http://localhost:8080"
+API_KEY = "your-api-key"  # 替换为实际的 API key
+
+def test_tool_call():
+    """测试工具调用功能"""
+    
+    # 定义一个简单的工具
+    tools = [
+        {
+            "type": "function",
+            "function": {
+                "name": "get_weather",
+                "description": "获取指定城市的天气信息",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "location": {
+                            "type": "string",
+                            "description": "城市名称，例如：北京、上海"
+                        },
+                        "unit": {
+                            "type": "string",
+                            "description": "温度单位",
+                            "enum": ["celsius", "fahrenheit"]
+                        }
+                    },
+                    "required": ["location"]
+                }
+            }
+        }
+    ]
+    
+    # 构建请求
+    request_data = {
+        "model": "GLM-4.5",
+        "messages": [
+            {
+                "role": "user",
+                "content": "北京的天气怎么样？"
+            }
+        ],
+        "tools": tools,
+        "tool_choice": "auto",
+        "stream": False
+    }
+    
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {API_KEY}"
+    }
+    
+    print("=" * 60)
+    print("测试工具调用 (非流式)")
+    print("=" * 60)
+    
+    # 发送请求
+    response = requests.post(
+        f"{BASE_URL}/v1/chat/completions",
+        json=request_data,
+        headers=headers
+    )
+    
+    print(f"状态码: {response.status_code}")
+    
+    if response.status_code == 200:
+        result = response.json()
+        print("\n响应内容:")
+        print(json.dumps(result, ensure_ascii=False, indent=2))
+        
+        # 检查是否有工具调用
+        if result.get("choices"):
+            choice = result["choices"][0]
+            if choice.get("message", {}).get("tool_calls"):
+                print("\n✅ 检测到工具调用!")
+                for tc in choice["message"]["tool_calls"]:
+                    print(f"  - 函数: {tc.get('function', {}).get('name')}")
+                    print(f"    参数: {tc.get('function', {}).get('arguments')}")
+            else:
+                print("\n⚠️ 未检测到工具调用")
+                if choice.get("message", {}).get("content"):
+                    print(f"内容: {choice['message']['content'][:200]}")
+    else:
+        print(f"\n错误响应: {response.text}")
+    
+    # 测试流式响应
+    print("\n" + "=" * 60)
+    print("测试工具调用 (流式)")
+    print("=" * 60)
+    
+    request_data["stream"] = True
+    
+    response = requests.post(
+        f"{BASE_URL}/v1/chat/completions",
+        json=request_data,
+        headers=headers,
+        stream=True
+    )
+    
+    print(f"状态码: {response.status_code}")
+    
+    if response.status_code == 200:
+        print("\n流式响应:")
+        tool_calls_detected = False
+        
+        for line in response.iter_lines():
+            if line:
+                line_str = line.decode('utf-8')
+                if line_str.startswith("data: "):
+                    data = line_str[6:]
+                    if data == "[DONE]":
+                        print("流结束")
+                        break
+                    
+                    try:
+                        chunk = json.loads(data)
+                        if chunk.get("choices"):
+                            delta = chunk["choices"][0].get("delta", {})
+                            if delta.get("tool_calls"):
+                                tool_calls_detected = True
+                                print(f"检测到工具调用: {json.dumps(delta['tool_calls'], ensure_ascii=False)}")
+                            elif delta.get("content"):
+                                print(f"内容: {delta['content']}", end="")
+                    except json.JSONDecodeError:
+                        pass
+        
+        if tool_calls_detected:
+            print("\n\n✅ 流式响应中检测到工具调用!")
+        else:
+            print("\n\n⚠️ 流式响应中未检测到工具调用")
+    else:
+        print(f"\n错误响应: {response.text}")
+
+
+if __name__ == "__main__":
+    test_tool_call()