Upload 81 files

Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.9-slim
+
+WORKDIR /code
+
+COPY ./requirements.txt /code/requirements.txt
+COPY ./api.py /code/api.py
+COPY ./json_parser.py /code/json_parser.py
+COPY ./logger_config.py /code/logger_config.py
+COPY ./response_formatter.py /code/response_formatter.py
+
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+EXPOSE 7860
+
+CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "7860"]

api.py

@@ -0,0 +1,346 @@
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from typing import Dict, List, Optional, Union, Any
+from pydantic import BaseModel, Field
+from datetime import datetime
+import logging
+import json
+import os
+from dotenv import load_dotenv
+from dify_client_python.dify_client import models
+from sse_starlette.sse import EventSourceResponse
+import httpx
+from json_parser import SSEParser
+from logger_config import setup_logger
+from fastapi.responses import StreamingResponse
+from fastapi.responses import JSONResponse
+from response_formatter import ResponseFormatter
+import traceback
+
+# Load environment variables
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+class AgentOutput(BaseModel):
+    """Structured output from agent processing"""
+    thought_content: str
+    observation: Optional[str]
+    tool_outputs: List[Dict]
+    citations: List[Dict]
+    metadata: Dict
+    raw_response: str
+
+class AgentRequest(BaseModel):
+    """Enhanced request model with additional parameters"""
+    query: str
+    conversation_id: Optional[str] = None
+    stream: bool = True
+    inputs: Dict = {}
+    files: List = []
+    user: str = "default_user"
+    response_mode: str = "streaming"
+
+class AgentProcessor:
+    def __init__(self, api_key: str):
+        self.api_key = api_key
+        self.api_base = "https://rag-engine.go-yamamoto.com/v1"
+        self.formatter = ResponseFormatter()
+        self.client = httpx.AsyncClient(timeout=60.0)
+        self.logger = setup_logger("agent_processor")
+
+    async def log_request_details(
+        self, 
+        request: AgentRequest,
+        start_time: datetime
+    ) -> None:
+        """Log detailed request information"""
+        self.logger.debug(
+            "Request details: \n"
+            f"Query: {request.query}\n"
+            f"User: {request.user}\n"
+            f"Conversation ID: {request.conversation_id}\n"
+            f"Stream mode: {request.stream}\n"
+            f"Start time: {start_time}\n"
+            f"Inputs: {request.inputs}\n"
+            f"Files: {len(request.files)} files attached"
+        )
+
+    async def log_error(
+        self, 
+        error: Exception,
+        context: Optional[Dict] = None
+    ) -> None:
+        """Log detailed error information"""
+        error_msg = (
+            f"Error type: {type(error).__name__}\n"
+            f"Error message: {str(error)}\n"
+            f"Stack trace:\n{traceback.format_exc()}\n"
+        )
+        if context:
+            error_msg += f"Context:\n{json.dumps(context, indent=2)}"
+        
+        self.logger.error(error_msg)
+
+    async def cleanup(self):
+        """Cleanup method to properly close client"""
+        await self.client.aclose()
+
+    async def process_stream(self, request: AgentRequest):
+        start_time = datetime.now()
+        await self.log_request_details(request, start_time)
+        
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+            "Accept": "text/event-stream"
+        }
+        
+        chat_request = {
+            "query": request.query,
+            "inputs": request.inputs,
+            "response_mode": "streaming" if request.stream else "blocking",
+            "user": request.user,
+            "conversation_id": request.conversation_id,
+            "files": request.files
+        }
+
+        async def event_generator():
+            parser = SSEParser()
+            citations = []
+            metadata = {}
+            
+            try:
+                async with self.client.stream(
+                    "POST",
+                    f"{self.api_base}/chat-messages",
+                    headers=headers,
+                    json=chat_request
+                ) as response:
+                    self.logger.debug(
+                        f"Stream connection established\n"
+                        f"Status: {response.status_code}\n"
+                        f"Headers: {dict(response.headers)}"
+                    )
+                    
+                    buffer = ""
+                    async for line in response.aiter_lines():
+                        if not line.strip():
+                            continue
+                        
+                        self.logger.debug(f"Raw SSE line: {line}")
+                        
+                        if "data:" in line:
+                            try:
+                                data = line.split("data:", 1)[1].strip()
+                                parsed = json.loads(data)
+                                
+                                if parsed.get("event") == "message_end":
+                                    citations = parsed.get("retriever_resources", [])
+                                    metadata = parsed.get("metadata", {})
+                                    self.logger.debug(
+                                        f"Message end event:\n"
+                                        f"Citations: {citations}\n"
+                                        f"Metadata: {metadata}"
+                                    )
+                                
+                                formatted = self.format_terminal_output(
+                                    parsed,
+                                    citations=citations,
+                                    metadata=metadata
+                                )
+                                if formatted:
+                                    self.logger.info(formatted)
+                            except Exception as e:
+                                await self.log_error(
+                                    e, 
+                                    {"line": line, "event": "parse_data"}
+                                )
+                        
+                        buffer += line + "\n"
+                        
+                        if line.startswith("data:") or buffer.strip().endswith("}"):
+                            try:
+                                processed_response = parser.parse_sse_event(buffer)
+                                if processed_response and isinstance(processed_response, dict):
+                                    cleaned_response = self.clean_response(processed_response)
+                                    if cleaned_response:
+                                        xml_content = cleaned_response.get("content", "")
+                                        yield f"data: {xml_content}\n\n"
+                            except Exception as parse_error:
+                                await self.log_error(
+                                    parse_error,
+                                    {"buffer": buffer, "event": "process_buffer"}
+                                )
+                                error_xml = (
+                                    f"<agent_response>"
+                                    f"<error>{str(parse_error)}</error>"
+                                    f"</agent_response>"
+                                )
+                                yield f"data: {error_xml}\n\n"
+                            finally:
+                                buffer = ""
+            
+            except httpx.ConnectError as e:
+                await self.log_error(e, {"event": "connection_error"})
+                error_xml = (
+                    f"<agent_response>"
+                    f"<error>Connection error: {str(e)}</error>"
+                    f"</agent_response>"
+                )
+                yield f"data: {error_xml}\n\n"
+            except Exception as e:
+                await self.log_error(e, {"event": "stream_error"})
+                error_xml = (
+                    f"<agent_response>"
+                    f"<error>Streaming error: {str(e)}</error>"
+                    f"</agent_response>"
+                )
+                yield f"data: {error_xml}\n\n"
+            finally:
+                end_time = datetime.now()
+                duration = (end_time - start_time).total_seconds()
+                self.logger.info(f"Request completed in {duration:.2f} seconds")
+
+        return StreamingResponse(
+            event_generator(),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+                "Access-Control-Allow-Origin": "*"
+            }
+        )
+
+    def format_terminal_output(
+        self, 
+        response: Dict, 
+        citations: List[Dict] = None,
+        metadata: Dict = None
+    ) -> Optional[str]:
+        """Format response for terminal output"""
+        event_type = response.get("event")
+        
+        if event_type == "agent_thought":
+            thought = response.get("thought", "")
+            observation = response.get("observation", "")
+            terminal_output, _ = self.formatter.format_thought(
+                thought, 
+                observation,
+                citations=citations,
+                metadata=metadata
+            )
+            return terminal_output
+            
+        elif event_type == "agent_message":
+            message = response.get("answer", "")
+            terminal_output, _ = self.formatter.format_message(message)
+            return terminal_output
+            
+        elif event_type == "error":
+            error = response.get("error", "Unknown error")
+            terminal_output, _ = self.formatter.format_error(error)
+            return terminal_output
+            
+        return None
+
+    def clean_response(self, response: Dict) -> Optional[Dict]:
+        """Clean and transform the response for frontend consumption"""
+        try:
+            event_type = response.get("event")
+            if not event_type:
+                return None
+
+            # Handle different event types
+            if event_type == "agent_thought":
+                thought = response.get("thought", "")
+                observation = response.get("observation", "")
+                _, xml_output = self.formatter.format_thought(thought, observation)
+                return {
+                    "type": "thought",
+                    "content": xml_output
+                }
+            
+            elif event_type == "agent_message":
+                message = response.get("answer", "")
+                _, xml_output = self.formatter.format_message(message)
+                return {
+                    "type": "message",
+                    "content": xml_output
+                }
+            
+            elif event_type == "error":
+                error = response.get("error", "Unknown error")
+                _, xml_output = self.formatter.format_error(error)
+                return {
+                    "type": "error",
+                    "content": xml_output
+                }
+            
+            return None
+        except Exception as e:
+            logger.error(f"Error cleaning response: {str(e)}")
+            return None
+
+# Initialize FastAPI app
+app = FastAPI()
+agent_processor = None
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.on_event("startup")
+async def startup_event():
+    global agent_processor
+    api_key = os.getenv("DIFY_API_KEY", "app-kVHTrZzEmFXEBfyXOi4rro7M")
+    agent_processor = AgentProcessor(api_key=api_key)
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    global agent_processor
+    if agent_processor:
+        await agent_processor.cleanup()
+
+@app.post("/v1/agent")
+async def process_agent_request(request: AgentRequest):
+    try:
+        logger.info(f"Processing agent request: {request.query}")
+        return await agent_processor.process_stream(request)
+        
+    except Exception as e:
+        logger.error(f"Error in agent request processing: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.middleware("http")
+async def error_handling_middleware(request: Request, call_next):
+    try:
+        response = await call_next(request)
+        return response
+    except Exception as e:
+        logger.error(f"Unhandled error: {str(e)}", exc_info=True)
+        return JSONResponse(
+            status_code=500,
+            content={"error": "Internal server error occurred"}
+        )
+
+# Add host and port parameters to the launch
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(
+        "api:app",
+        host="0.0.0.0",
+        port=8224,
+        reload=True
+    )

api_docs.md

@@ -0,0 +1,154 @@
+POST
+/chat-messages
+Send Chat Message
+Send a request to the chat application.
+
+Request Body
+Name
+query
+Type
+string
+Description
+User Input/Question content
+
+Name
+inputs
+Type
+object
+Description
+Allows the entry of various variable values defined by the App. The inputs parameter contains multiple key/value pairs, with each key corresponding to a specific variable and each value being the specific value for that variable. Default {}
+
+Name
+response_mode
+Type
+string
+Description
+The mode of response return, supporting:
+
+streaming Streaming mode (recommended), implements a typewriter-like output through SSE (Server-Sent Events).
+blocking Blocking mode, returns result after execution is complete. (Requests may be interrupted if the process is long) Due to Cloudflare restrictions, the request will be interrupted without a return after 100 seconds. Note: blocking mode is not supported in Agent Assistant mode
+Name
+user
+Type
+string
+Description
+User identifier, used to define the identity of the end-user for retrieval and statistics. Should be uniquely defined by the developer within the application.
+
+Name
+conversation_id
+Type
+string
+Description
+Conversation ID, to continue the conversation based on previous chat records, it is necessary to pass the previous message's conversation_id.
+
+Name
+files
+Type
+array[object]
+Description
+File list, suitable for inputting files (images) combined with text understanding and answering questions, available only when the model supports Vision capability.
+
+type (string) Supported type: image (currently only supports image type)
+transfer_method (string) Transfer method, remote_url for image URL / local_file for file upload
+url (string) Image URL (when the transfer method is remote_url)
+upload_file_id (string) Uploaded file ID, which must be obtained by uploading through the File Upload API in advance (when the transfer method is local_file)
+Name
+auto_generate_name
+Type
+bool
+Description
+Auto-generate title, default is true. If set to false, can achieve async title generation by calling the conversation rename API and setting auto_generate to true.
+
+Response
+When response_mode is blocking, return a CompletionResponse object. When response_mode is streaming, return a ChunkCompletionResponse stream.
+
+ChatCompletionResponse
+Returns the complete App result, Content-Type is application/json.
+
+message_id (string) Unique message ID
+conversation_id (string) Conversation ID
+mode (string) App mode, fixed as chat
+answer (string) Complete response content
+metadata (object) Metadata
+usage (Usage) Model usage information
+retriever_resources (array[RetrieverResource]) Citation and Attribution List
+created_at (int) Message creation timestamp, e.g., 1705395332
+ChunkChatCompletionResponse
+Returns the stream chunks outputted by the App, Content-Type is text/event-stream. Each streaming chunk starts with data:, separated by two newline characters \n\n, as shown below:
+
+data: {"event": "message", "task_id": "900bbd43-dc0b-4383-a372-aa6e6c414227", "id": "663c5084-a254-4040-8ad3-51f2a3c1a77c", "answer": "Hi", "created_at": 1705398420}\n\n
+
+Copy
+Copied!
+The structure of the streaming chunks varies depending on the event:
+
+event: message LLM returns text chunk event, i.e., the complete text is output in a chunked fashion.
+task_id (string) Task ID, used for request tracking and the below Stop Generate API
+message_id (string) Unique message ID
+conversation_id (string) Conversation ID
+answer (string) LLM returned text chunk content
+created_at (int) Creation timestamp, e.g., 1705395332
+event: agent_message LLM returns text chunk event, i.e., with Agent Assistant enabled, the complete text is output in a chunked fashion (Only supported in Agent mode)
+task_id (string) Task ID, used for request tracking and the below Stop Generate API
+message_id (string) Unique message ID
+conversation_id (string) Conversation ID
+answer (string) LLM returned text chunk content
+created_at (int) Creation timestamp, e.g., 1705395332
+event: tts_message TTS audio stream event, that is, speech synthesis output. The content is an audio block in Mp3 format, encoded as a base64 string. When playing, simply decode the base64 and feed it into the player. (This message is available only when auto-play is enabled)
+task_id (string) Task ID, used for request tracking and the stop response interface below
+message_id (string) Unique message ID
+audio (string) The audio after speech synthesis, encoded in base64 text content, when playing, simply decode the base64 and feed it into the player
+created_at (int) Creation timestamp, e.g.: 1705395332
+event: tts_message_end TTS audio stream end event, receiving this event indicates the end of the audio stream.
+task_id (string) Task ID, used for request tracking and the stop response interface below
+message_id (string) Unique message ID
+audio (string) The end event has no audio, so this is an empty string
+created_at (int) Creation timestamp, e.g.: 1705395332
+event: agent_thought thought of Agent, contains the thought of LLM, input and output of tool calls (Only supported in Agent mode)
+id (string) Agent thought ID, every iteration has a unique agent thought ID
+task_id (string) (string) Task ID, used for request tracking and the below Stop Generate API
+message_id (string) Unique message ID
+position (int) Position of current agent thought, each message may have multiple thoughts in order.
+thought (string) What LLM is thinking about
+observation (string) Response from tool calls
+tool (string) A list of tools represents which tools are called，split by ;
+tool_input (string) Input of tools in JSON format. Like: {"dalle3": {"prompt": "a cute cat"}}.
+created_at (int) Creation timestamp, e.g., 1705395332
+message_files (array[string]) Refer to message_file event
+file_id (string) File ID
+conversation_id (string) Conversation ID
+event: message_file Message file event, a new file has created by tool
+id (string) File unique ID
+type (string) File type，only allow "image" currently
+belongs_to (string) Belongs to, it will only be an 'assistant' here
+url (string) Remote url of file
+conversation_id (string) Conversation ID
+event: message_end Message end event, receiving this event means streaming has ended.
+task_id (string) Task ID, used for request tracking and the below Stop Generate API
+message_id (string) Unique message ID
+conversation_id (string) Conversation ID
+metadata (object) Metadata
+usage (Usage) Model usage information
+retriever_resources (array[RetrieverResource]) Citation and Attribution List
+event: message_replace Message content replacement event. When output content moderation is enabled, if the content is flagged, then the message content will be replaced with a preset reply through this event.
+task_id (string) Task ID, used for request tracking and the below Stop Generate API
+message_id (string) Unique message ID
+conversation_id (string) Conversation ID
+answer (string) Replacement content (directly replaces all LLM reply text)
+created_at (int) Creation timestamp, e.g., 1705395332
+event: error Exceptions that occur during the streaming process will be output in the form of stream events, and reception of an error event will end the stream.
+task_id (string) Task ID, used for request tracking and the below Stop Generate API
+message_id (string) Unique message ID
+status (int) HTTP status code
+code (string) Error code
+message (string) Error message
+event: ping Ping event every 10 seconds to keep the connection alive.
+Errors
+404, Conversation does not exists
+400, invalid_param, abnormal parameter input
+400, app_unavailable, App configuration unavailable
+400, provider_not_initialize, no available model credential configuration
+400, provider_quota_exceeded, model invocation quota insufficient
+400, model_currently_not_support, current model unavailable
+400, completion_request_error, text generation failed
+500, internal server error

dify_client_python/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 haoyuhu
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

dify_client_python/MANIFEST.in

@@ -0,0 +1 @@
+recursive-include dify_client *.py

dify_client_python/README.md

@@ -0,0 +1,155 @@
+# dify-client-python
+
+Welcome to the `dify-client-python` repository! This Python package provides a convenient and powerful interface to
+interact with the Dify API, enabling developers to integrate a wide range of features into their applications with ease.
+
+## Main Features
+
+* **Synchronous and Asynchronous Support**: The client offers both synchronous and asynchronous methods, allowing for
+  flexible integration into various Python codebases and frameworks.
+* **Stream and Non-stream Support**: Seamlessly work with both streaming and non-streaming endpoints of the Dify API for
+  real-time and batch processing use cases.
+* **Comprehensive Endpoint Coverage**: Support completion, chat, workflows, feedback, file uploads, etc., the client
+  covers all available Dify API endpoints.
+
+## Installation
+
+Before using the `dify-client-python` client, you'll need to install it. You can easily install it using `pip`:
+
+```bash
+pip install dify-client-python
+```
+
+## Quick Start
+
+Here's a quick example of how you can use the Dify Client to send a chat message.
+
+```python
+import uuid
+from dify_client import Client, models
+
+# Initialize the client with your API key
+client = Client(
+    api_key="your-api-key",
+    api_base="http://localhost/v1",
+)
+user = str(uuid.uuid4())
+
+# Create a blocking chat request
+blocking_chat_req = models.ChatRequest(
+    query="Hi, dify-client-python!",
+    inputs={"city": "Beijing"},
+    user=user,
+    response_mode=models.ResponseMode.BLOCKING,
+)
+
+# Send the chat message
+chat_response = client.chat_messages(blocking_chat_req, timeout=60.)
+print(chat_response)
+
+# Create a streaming chat request
+streaming_chat_req = models.ChatRequest(
+    query="Hi, dify-client-python!",
+    inputs={"city": "Beijing"},
+    user=user,
+    response_mode=models.ResponseMode.STREAMING,
+)
+
+# Send the chat message
+for chunk in client.chat_messages(streaming_chat_req, timeout=60.):
+    print(chunk)
+```
+
+For asynchronous operations, use the `AsyncClient` in a similar fashion:
+
+```python
+import asyncio
+import uuid
+
+from dify_client import AsyncClient, models
+
+# Initialize the async client with your API key
+async_client = AsyncClient(
+    api_key="your-api-key",
+    api_base="http://localhost/v1",
+)
+
+
+# Define an asynchronous function to send a blocking chat message with BLOCKING ResponseMode
+async def send_chat_message():
+    user = str(uuid.uuid4())
+    # Create a blocking chat request
+    blocking_chat_req = models.ChatRequest(
+        query="Hi, dify-client-python!",
+        inputs={"city": "Beijing"},
+        user=user,
+        response_mode=models.ResponseMode.BLOCKING,
+    )
+    chat_response = await async_client.achat_messages(blocking_chat_req, timeout=60.)
+    print(chat_response)
+
+
+# Define an asynchronous function to send a chat message with STREAMING ResponseMode
+async def send_chat_message_stream():
+    user = str(uuid.uuid4())
+    # Create a blocking chat request
+    streaming_chat_req = models.ChatRequest(
+        query="Hi, dify-client-python!",
+        inputs={"city": "Beijing"},
+        user=user,
+        response_mode=models.ResponseMode.STREAMING,
+    )
+    async for chunk in await async_client.achat_messages(streaming_chat_req, timeout=60.):
+        print(chunk)
+
+
+# Run the asynchronous function
+asyncio.gather(send_chat_message(), send_chat_message_stream())
+```
+
+## Documentation
+
+For detailed information on all the functionalities and how to use each endpoint, please refer to the official Dify API
+documentation. This will provide you with comprehensive guidance on request and response structures, error handling, and
+other important details.
+
+## Contributing
+
+Contributions are welcome! If you would like to contribute to the `dify-client-python`, please feel free to make a pull
+request or open an issue to discuss potential changes.
+
+## License
+
+This project is licensed under the MIT License - see the LICENSE file for details.
+
+```text
+MIT License
+
+Copyright (c) 2024 haoyuhu
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
+```
+
+## Support
+
+If you encounter any issues or have questions regarding the usage of this client, please reach out to the Dify Client
+support team.
+
+Happy coding! 🚀

dify_client_python/build.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+
+set -e
+
+rm -rf build dist *.egg-info
+
+pip install setuptools wheel twine
+python setup.py sdist bdist_wheel
+twine upload dist/*

dify_client_python/build/lib/dify_client/__init__.py

@@ -0,0 +1 @@
+from ._clientx import Client, AsyncClient

dify_client_python/build/lib/dify_client/_clientx.py

@@ -0,0 +1,660 @@
+from typing import Optional, Any, Mapping, Iterator, AsyncIterator, Union, Dict
+
+try:
+    from enum import StrEnum
+except ImportError:
+    from strenum import StrEnum
+try:
+    from http import HTTPMethod
+except ImportError:
+    class HTTPMethod(StrEnum):
+        GET = "GET"
+        POST = "POST"
+        PUT = "PUT"
+        DELETE = "DELETE"
+
+import httpx
+# noinspection PyProtectedMember
+import httpx._types as types
+from httpx_sse import connect_sse, ServerSentEvent, aconnect_sse
+from pydantic import BaseModel
+
+from dify_client import errors, models
+
+_httpx_client = httpx.Client()
+_async_httpx_client = httpx.AsyncClient()
+
+IGNORED_STREAM_EVENTS = (models.StreamEvent.PING.value,)
+
+# feedback
+ENDPOINT_FEEDBACKS = "/messages/{message_id}/feedbacks"
+# suggest
+ENDPOINT_SUGGESTED = "/messages/{message_id}/suggested"
+# files upload
+ENDPOINT_FILES_UPLOAD = "/files/upload"
+# completion
+ENDPOINT_COMPLETION_MESSAGES = "/completion-messages"
+ENDPOINT_STOP_COMPLETION_MESSAGES = "/completion-messages/{task_id}/stop"
+# chat
+ENDPOINT_CHAT_MESSAGES = "/chat-messages"
+ENDPOINT_STOP_CHAT_MESSAGES = "/chat-messages/{task_id}/stop"
+# workflow
+ENDPOINT_RUN_WORKFLOWS = "/workflows/run"
+ENDPOINT_STOP_WORKFLOWS = "/workflows/{task_id}/stop"
+# audio <-> text
+ENDPOINT_TEXT_TO_AUDIO = "/text-to-audio"
+ENDPOINT_AUDIO_TO_TEXT = "/audio-to-text"
+
+
+class Client(BaseModel):
+    api_key: str
+    api_base: Optional[str] = "https://api.dify.ai/v1"
+
+    def request(self, endpoint: str, method: str,
+                content: Optional[types.RequestContent] = None,
+                data: Optional[types.RequestData] = None,
+                files: Optional[types.RequestFiles] = None,
+                json: Optional[Any] = None,
+                params: Optional[types.QueryParamTypes] = None,
+                headers: Optional[Mapping[str, str]] = None,
+                **kwargs: object,
+                ) -> httpx.Response:
+        """
+        Sends a synchronous HTTP request to the specified endpoint.
+
+        Args:
+            endpoint: The API endpoint to send the request to.
+            method: The HTTP method to use (e.g., 'GET', 'POST').
+            content: Raw content to include in the request body.
+            data: Form data to include in the request body.
+            files: Files to include in the request body.
+            json: JSON data to include in the request body.
+            params: Query parameters to include in the request URL.
+            headers: Additional headers to include in the request.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `httpx.Response` object containing the HTTP response.
+
+        Raises:
+            Various DifyAPIError exceptions if the response contains an error.
+        """
+        merged_headers = {}
+        if headers:
+            merged_headers.update(headers)
+        self._prepare_auth_headers(merged_headers)
+
+        response = _httpx_client.request(method, endpoint, content=content, data=data, files=files, json=json,
+                                         params=params, headers=merged_headers, **kwargs)
+        errors.raise_for_status(response)
+        return response
+
+    def request_stream(self, endpoint: str, method: str,
+                       content: Optional[types.RequestContent] = None,
+                       data: Optional[types.RequestData] = None,
+                       files: Optional[types.RequestFiles] = None,
+                       json: Optional[Any] = None,
+                       params: Optional[types.QueryParamTypes] = None,
+                       headers: Optional[Mapping[str, str]] = None,
+                       **kwargs,
+                       ) -> Iterator[ServerSentEvent]:
+        """
+        Opens a server-sent events (SSE) stream to the specified endpoint.
+
+        Args:
+            endpoint: The API endpoint to send the request to.
+            method: The HTTP method to use (e.g., 'GET', 'POST').
+            content: Raw content to include in the request body.
+            data: Form data to include in the request body.
+            files: Files to include in the request body.
+            json: JSON data to include in the request body.
+            params: Query parameters to include in the request URL.
+            headers: Additional headers to include in the request.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            An iterator of `ServerSentEvent` objects representing the stream of events.
+
+        Raises:
+            Various DifyAPIError exceptions if an error event is received in the stream.
+        """
+        merged_headers = {}
+        if headers:
+            merged_headers.update(headers)
+        self._prepare_auth_headers(merged_headers)
+
+        with connect_sse(_httpx_client, method, endpoint, headers=merged_headers,
+                         content=content, data=data, files=files, json=json, params=params, **kwargs) as event_source:
+            if not _check_stream_content_type(event_source.response):
+                event_source.response.read()
+                errors.raise_for_status(event_source.response)
+            for sse in event_source.iter_sse():
+                errors.raise_for_status(sse)
+                if sse.event in IGNORED_STREAM_EVENTS or sse.data in IGNORED_STREAM_EVENTS:
+                    continue
+                yield sse
+
+    def feedback_messages(self, message_id: str, req: models.FeedbackRequest, **kwargs) -> models.FeedbackResponse:
+        """
+        Submits feedback for a specific message.
+
+        Args:
+            message_id: The identifier of the message to submit feedback for.
+            req: A `FeedbackRequest` object containing the feedback details, such as the rating.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `FeedbackResponse` object containing the result of the feedback submission.
+        """
+        response = self.request(
+            self._prepare_url(ENDPOINT_FEEDBACKS, message_id=message_id),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.FeedbackResponse(**response.json())
+
+    def suggest_messages(self, message_id: str, req: models.ChatSuggestRequest, **kwargs) -> models.ChatSuggestResponse:
+        """
+        Retrieves suggested messages based on a specific message.
+
+        Args:
+            message_id: The identifier of the message to get suggestions for.
+            req: A `ChatSuggestRequest` object containing the request details.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `ChatSuggestResponse` object containing suggested messages.
+        """
+        response = self.request(
+            self._prepare_url(ENDPOINT_SUGGESTED, message_id=message_id),
+            HTTPMethod.GET,
+            params=req.model_dump(),
+            **kwargs,
+        )
+        return models.ChatSuggestResponse(**response.json())
+
+    def upload_files(self, file: types.FileTypes, req: models.UploadFileRequest,
+                     **kwargs) -> models.UploadFileResponse:
+        """
+        Uploads a file to be used in subsequent requests.
+
+        Args:
+            file: The file to upload. This can be a file-like object, or a tuple of
+            (`filename`, file-like object, mime_type).
+            req: An `UploadFileRequest` object containing the upload details, such as the user who is uploading.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            An `UploadFileResponse` object containing details about the uploaded file, such as its identifier and URL.
+        """
+        response = self.request(
+            self._prepare_url(ENDPOINT_FILES_UPLOAD),
+            HTTPMethod.POST,
+            data=req.model_dump(),
+            files=[("file", file)],
+            **kwargs,
+        )
+        return models.UploadFileResponse(**response.json())
+
+    def completion_messages(self, req: models.CompletionRequest, **kwargs) \
+            -> Union[models.CompletionResponse, Iterator[models.CompletionStreamResponse]]:
+        """
+        Sends a request to generate a completion or a series of completions based on the provided input.
+
+        Returns:
+            If the response mode is blocking, it returns a `CompletionResponse` object containing the generated message.
+            If the response mode is streaming, it returns an iterator of `CompletionStreamResponse` objects containing
+            the stream of generated events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return self._completion_messages(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._completion_messages_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    def _completion_messages(self, req: models.CompletionRequest, **kwargs) -> models.CompletionResponse:
+        response = self.request(
+            self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.CompletionResponse(**response.json())
+
+    def _completion_messages_stream(self, req: models.CompletionRequest, **kwargs) \
+            -> Iterator[models.CompletionStreamResponse]:
+        event_source = self.request_stream(
+            self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        for sse in event_source:
+            yield models.build_completion_stream_response(sse.json())
+
+    def stop_completion_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming completion task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return self._stop_stream(self._prepare_url(ENDPOINT_STOP_COMPLETION_MESSAGES, task_id=task_id), req, **kwargs)
+
+    def chat_messages(self, req: models.ChatRequest, **kwargs) \
+            -> Union[models.ChatResponse, Iterator[models.ChatStreamResponse]]:
+        """
+        Sends a request to generate a chat message or a series of chat messages based on the provided input.
+
+        Returns:
+            If the response mode is blocking, it returns a `ChatResponse` object containing the generated chat message.
+            If the response mode is streaming, it returns an iterator of `ChatStreamResponse` objects containing the
+            stream of chat events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return self._chat_messages(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._chat_messages_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    def _chat_messages(self, req: models.ChatRequest, **kwargs) -> models.ChatResponse:
+        response = self.request(
+            self._prepare_url(ENDPOINT_CHAT_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.ChatResponse(**response.json())
+
+    def _chat_messages_stream(self, req: models.ChatRequest, **kwargs) -> Iterator[models.ChatStreamResponse]:
+        event_source = self.request_stream(
+            self._prepare_url(ENDPOINT_CHAT_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        for sse in event_source:
+            yield models.build_chat_stream_response(sse.json())
+
+    def stop_chat_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming chat task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return self._stop_stream(self._prepare_url(ENDPOINT_STOP_CHAT_MESSAGES, task_id=task_id), req, **kwargs)
+
+    def run_workflows(self, req: models.WorkflowsRunRequest, **kwargs) \
+            -> Union[models.WorkflowsRunResponse, Iterator[models.WorkflowsRunStreamResponse]]:
+        """
+        Initiates the execution of a workflow, which can consist of multiple steps and actions.
+
+        Returns:
+            If the response mode is blocking, it returns a `WorkflowsRunResponse` object containing the results of the
+            completed workflow.
+            If the response mode is streaming, it returns an iterator of `WorkflowsRunStreamResponse` objects
+            containing the stream of workflow events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return self._run_workflows(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._run_workflows_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    def _run_workflows(self, req: models.WorkflowsRunRequest, **kwargs) -> models.WorkflowsRunResponse:
+        response = self.request(
+            self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.WorkflowsRunResponse(**response.json())
+
+    def _run_workflows_stream(self, req: models.WorkflowsRunRequest, **kwargs) \
+            -> Iterator[models.WorkflowsRunStreamResponse]:
+        event_source = self.request_stream(
+            self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        for sse in event_source:
+            yield models.build_workflows_stream_response(sse.json())
+
+    def stop_workflows(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming workflow task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return self._stop_stream(self._prepare_url(ENDPOINT_STOP_WORKFLOWS, task_id=task_id), req, **kwargs)
+
+    def _stop_stream(self, endpoint: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        response = self.request(
+            endpoint,
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.StopResponse(**response.json())
+
+    def _prepare_url(self, endpoint: str, **kwargs) -> str:
+        return self.api_base + endpoint.format(**kwargs)
+
+    def _prepare_auth_headers(self, headers: Dict[str, str]):
+        if "authorization" not in (key.lower() for key in headers.keys()):
+            headers["Authorization"] = f"Bearer {self.api_key}"
+
+
+class AsyncClient(BaseModel):
+    api_key: str
+    api_base: Optional[str] = "https://api.dify.ai/v1"
+
+    async def arequest(self, endpoint: str, method: str,
+                       content: Optional[types.RequestContent] = None,
+                       data: Optional[types.RequestData] = None,
+                       files: Optional[types.RequestFiles] = None,
+                       json: Optional[Any] = None,
+                       params: Optional[types.QueryParamTypes] = None,
+                       headers: Optional[Mapping[str, str]] = None,
+                       **kwargs,
+                       ) -> httpx.Response:
+        """
+        Asynchronously sends a request to the specified Dify API endpoint.
+
+        Args:
+            endpoint: The endpoint URL to which the request is sent.
+            method: The HTTP method to be used for the request (e.g., 'GET', 'POST').
+            content: Raw content to include in the request body, if any.
+            data: Form data to be sent in the request body.
+            files: Files to be uploaded with the request.
+            json: JSON data to be sent in the request body.
+            params: Query parameters to be included in the request URL.
+            headers: Additional headers to be sent with the request.
+            **kwargs: Extra keyword arguments to be passed to the underlying HTTPX request function.
+
+        Returns:
+            A httpx.Response object containing the server's response to the HTTP request.
+
+        Raises:
+            Various DifyAPIError exceptions if the response contains an error.
+        """
+        merged_headers = {}
+        if headers:
+            merged_headers.update(headers)
+        self._prepare_auth_headers(merged_headers)
+
+        response = await _async_httpx_client.request(method, endpoint, content=content, data=data, files=files,
+                                                     json=json, params=params, headers=merged_headers, **kwargs)
+        errors.raise_for_status(response)
+        return response
+
+    async def arequest_stream(self, endpoint: str, method: str,
+                              content: Optional[types.RequestContent] = None,
+                              data: Optional[types.RequestData] = None,
+                              files: Optional[types.RequestFiles] = None,
+                              json: Optional[Any] = None,
+                              params: Optional[types.QueryParamTypes] = None,
+                              headers: Optional[Mapping[str, str]] = None,
+                              **kwargs,
+                              ) -> AsyncIterator[ServerSentEvent]:
+        """
+        Asynchronously establishes a streaming connection to the specified Dify API endpoint.
+
+        Args:
+            endpoint: The endpoint URL to which the request is sent.
+            method: The HTTP method to be used for the request (e.g., 'GET', 'POST').
+            content: Raw content to include in the request body, if any.
+            data: Form data to be sent in the request body.
+            files: Files to be uploaded with the request.
+            json: JSON data to be sent in the request body.
+            params: Query parameters to be included in the request URL.
+            headers: Additional headers to be sent with the request.
+            **kwargs: Extra keyword arguments to be passed to the underlying HTTPX request function.
+
+        Yields:
+            ServerSentEvent objects representing the events received from the server.
+
+        Raises:
+            Various DifyAPIError exceptions if an error event is received in the stream.
+        """
+        merged_headers = {}
+        if headers:
+            merged_headers.update(headers)
+        self._prepare_auth_headers(merged_headers)
+
+        async with aconnect_sse(_async_httpx_client, method, endpoint, headers=merged_headers,
+                                content=content, data=data, files=files, json=json, params=params,
+                                **kwargs) as event_source:
+            if not _check_stream_content_type(event_source.response):
+                await event_source.response.aread()
+                errors.raise_for_status(event_source.response)
+            async for sse in event_source.aiter_sse():
+                errors.raise_for_status(sse)
+                if sse.event in IGNORED_STREAM_EVENTS or sse.data in IGNORED_STREAM_EVENTS:
+                    continue
+                yield sse
+
+    async def afeedback_messages(self, message_id: str, req: models.FeedbackRequest, **kwargs) \
+            -> models.FeedbackResponse:
+        """
+        Submits feedback for a specific message.
+
+        Args:
+            message_id: The identifier of the message to submit feedback for.
+            req: A `FeedbackRequest` object containing the feedback details, such as the rating.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `FeedbackResponse` object containing the result of the feedback submission.
+        """
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_FEEDBACKS, message_id=message_id),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.FeedbackResponse(**response.json())
+
+    async def asuggest_messages(self, message_id: str, req: models.ChatSuggestRequest, **kwargs) \
+            -> models.ChatSuggestResponse:
+        """
+        Retrieves suggested messages based on a specific message.
+
+        Args:
+            message_id: The identifier of the message to get suggestions for.
+            req: A `ChatSuggestRequest` object containing the request details.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `ChatSuggestResponse` object containing suggested messages.
+        """
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_SUGGESTED, message_id=message_id),
+            HTTPMethod.GET,
+            params=req.model_dump(),
+            **kwargs,
+        )
+        return models.ChatSuggestResponse(**response.json())
+
+    async def aupload_files(self, file: types.FileTypes, req: models.UploadFileRequest, **kwargs) \
+            -> models.UploadFileResponse:
+        """
+        Uploads a file to be used in subsequent requests.
+
+        Args:
+            file: The file to upload. This can be a file-like object, or a tuple of
+            (`filename`, file-like object, mime_type).
+            req: An `UploadFileRequest` object containing the upload details, such as the user who is uploading.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            An `UploadFileResponse` object containing details about the uploaded file, such as its identifier and URL.
+        """
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_FILES_UPLOAD),
+            HTTPMethod.POST,
+            data=req.model_dump(),
+            files=[("file", file)],
+            **kwargs,
+        )
+        return models.UploadFileResponse(**response.json())
+
+    async def acompletion_messages(self, req: models.CompletionRequest, **kwargs) \
+            -> Union[models.CompletionResponse, AsyncIterator[models.CompletionStreamResponse]]:
+        """
+        Sends a request to generate a completion or a series of completions based on the provided input.
+
+        Returns:
+            If the response mode is blocking, it returns a `CompletionResponse` object containing the generated message.
+            If the response mode is streaming, it returns an iterator of `CompletionStreamResponse` objects containing
+            the stream of generated events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return await self._acompletion_messages(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._acompletion_messages_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    async def _acompletion_messages(self, req: models.CompletionRequest, **kwargs) -> models.CompletionResponse:
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.CompletionResponse(**response.json())
+
+    async def _acompletion_messages_stream(self, req: models.CompletionRequest, **kwargs) \
+            -> AsyncIterator[models.CompletionStreamResponse]:
+        async for sse in self.arequest_stream(
+                self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
+                HTTPMethod.POST,
+                json=req.model_dump(),
+                **kwargs):
+            yield models.build_completion_stream_response(sse.json())
+
+    async def astop_completion_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming completion task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return await self._astop_stream(
+            self._prepare_url(ENDPOINT_STOP_COMPLETION_MESSAGES, task_id=task_id), req, **kwargs)
+
+    async def achat_messages(self, req: models.ChatRequest, **kwargs) \
+            -> Union[models.ChatResponse, AsyncIterator[models.ChatStreamResponse]]:
+        """
+        Sends a request to generate a chat message or a series of chat messages based on the provided input.
+
+        Returns:
+            If the response mode is blocking, it returns a `ChatResponse` object containing the generated chat message.
+            If the response mode is streaming, it returns an iterator of `ChatStreamResponse` objects containing the
+            stream of chat events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return await self._achat_messages(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._achat_messages_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    async def _achat_messages(self, req: models.ChatRequest, **kwargs) -> models.ChatResponse:
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_CHAT_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.ChatResponse(**response.json())
+
+    async def _achat_messages_stream(self, req: models.ChatRequest, **kwargs) \
+            -> AsyncIterator[models.ChatStreamResponse]:
+        async for sse in self.arequest_stream(
+                self._prepare_url(ENDPOINT_CHAT_MESSAGES),
+                HTTPMethod.POST,
+                json=req.model_dump(),
+                **kwargs):
+            yield models.build_chat_stream_response(sse.json())
+
+    async def astop_chat_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming chat task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return await self._astop_stream(self._prepare_url(ENDPOINT_STOP_CHAT_MESSAGES, task_id=task_id), req, **kwargs)
+
+    async def arun_workflows(self, req: models.WorkflowsRunRequest, **kwargs) \
+            -> Union[models.WorkflowsRunResponse, AsyncIterator[models.WorkflowsStreamResponse]]:
+        """
+        Initiates the execution of a workflow, which can consist of multiple steps and actions.
+
+        Returns:
+            If the response mode is blocking, it returns a `WorkflowsRunResponse` object containing the results of the
+            completed workflow.
+            If the response mode is streaming, it returns an iterator of `WorkflowsRunStreamResponse` objects
+            containing the stream of workflow events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return await self._arun_workflows(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._arun_workflows_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    async def _arun_workflows(self, req: models.WorkflowsRunRequest, **kwargs) -> models.WorkflowsRunResponse:
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.WorkflowsRunResponse(**response.json())
+
+    async def _arun_workflows_stream(self, req: models.WorkflowsRunRequest, **kwargs) \
+            -> AsyncIterator[models.WorkflowsRunStreamResponse]:
+        async for sse in self.arequest_stream(
+                self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
+                HTTPMethod.POST,
+                json=req.model_dump(),
+                **kwargs):
+            yield models.build_workflows_stream_response(sse.json())
+
+    async def astop_workflows(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming workflow task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return await self._astop_stream(self._prepare_url(ENDPOINT_STOP_WORKFLOWS, task_id=task_id), req, **kwargs)
+
+    async def _astop_stream(self, endpoint: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        response = await self.arequest(
+            endpoint,
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.StopResponse(**response.json())
+
+    def _prepare_url(self, endpoint: str, **kwargs) -> str:
+        return self.api_base + endpoint.format(**kwargs)
+
+    def _prepare_auth_headers(self, headers: Dict[str, str]):
+        if "authorization" not in (key.lower() for key in headers.keys()):
+            headers["Authorization"] = f"Bearer {self.api_key}"
+
+
+def _get_content_type(headers: httpx.Headers) -> str:
+    return headers.get("content-type", "").partition(";")[0]
+
+
+def _check_stream_content_type(response: httpx.Response) -> bool:
+    content_type = _get_content_type(response.headers)
+    return response.is_success and "text/event-stream" in content_type

dify_client_python/build/lib/dify_client/errors.py

@@ -0,0 +1,132 @@
+from http import HTTPStatus
+from typing import Union
+
+import httpx
+import httpx_sse
+
+from dify_client import models
+
+
+class DifyAPIError(Exception):
+    def __init__(self, status: int, code: str, message: str):
+        super().__init__(f"status_code={status}, code={code}, {message}")
+        self.status = status
+        self.code = code
+        self.message = message
+
+
+class DifyInvalidParam(DifyAPIError):
+    pass
+
+
+class DifyNotChatApp(DifyAPIError):
+    pass
+
+
+class DifyResourceNotFound(DifyAPIError):
+    pass
+
+
+class DifyAppUnavailable(DifyAPIError):
+    pass
+
+
+class DifyProviderNotInitialize(DifyAPIError):
+    pass
+
+
+class DifyProviderQuotaExceeded(DifyAPIError):
+    pass
+
+
+class DifyModelCurrentlyNotSupport(DifyAPIError):
+    pass
+
+
+class DifyCompletionRequestError(DifyAPIError):
+    pass
+
+
+class DifyInternalServerError(DifyAPIError):
+    pass
+
+
+class DifyNoFileUploaded(DifyAPIError):
+    pass
+
+
+class DifyTooManyFiles(DifyAPIError):
+    pass
+
+
+class DifyUnsupportedPreview(DifyAPIError):
+    pass
+
+
+class DifyUnsupportedEstimate(DifyAPIError):
+    pass
+
+
+class DifyFileTooLarge(DifyAPIError):
+    pass
+
+
+class DifyUnsupportedFileType(DifyAPIError):
+    pass
+
+
+class DifyS3ConnectionFailed(DifyAPIError):
+    pass
+
+
+class DifyS3PermissionDenied(DifyAPIError):
+    pass
+
+
+class DifyS3FileTooLarge(DifyAPIError):
+    pass
+
+
+SPEC_CODE_ERRORS = {
+    # completion & chat & workflow
+    "invalid_param": DifyInvalidParam,
+    "not_chat_app": DifyNotChatApp,
+    "app_unavailable": DifyAppUnavailable,
+    "provider_not_initialize": DifyProviderNotInitialize,
+    "provider_quota_exceeded": DifyProviderQuotaExceeded,
+    "model_currently_not_support": DifyModelCurrentlyNotSupport,
+    "completion_request_error": DifyCompletionRequestError,
+    # files upload
+    "no_file_uploaded": DifyNoFileUploaded,
+    "too_many_files": DifyTooManyFiles,
+    "unsupported_preview": DifyUnsupportedPreview,
+    "unsupported_estimate": DifyUnsupportedEstimate,
+    "file_too_large": DifyFileTooLarge,
+    "unsupported_file_type": DifyUnsupportedFileType,
+    "s3_connection_failed": DifyS3ConnectionFailed,
+    "s3_permission_denied": DifyS3PermissionDenied,
+    "s3_file_too_large": DifyS3FileTooLarge,
+}
+
+
+def raise_for_status(response: Union[httpx.Response, httpx_sse.ServerSentEvent]):
+    if isinstance(response, httpx.Response):
+        if response.is_success:
+            return
+        json = response.json()
+        if "status" not in json:
+            json["status"] = response.status_code
+        details = models.ErrorResponse(**json)
+    elif isinstance(response, httpx_sse.ServerSentEvent):
+        if response.event != models.StreamEvent.ERROR.value:
+            return
+        details = models.ErrorStreamResponse(**response.json())
+    else:
+        raise ValueError(f"Invalid dify response type: {type(response)}")
+
+    if details.status == HTTPStatus.NOT_FOUND:
+        raise DifyResourceNotFound(details.status, details.code, details.message)
+    elif details.status == HTTPStatus.INTERNAL_SERVER_ERROR:
+        raise DifyInternalServerError(details.status, details.code, details.message)
+    else:
+        raise SPEC_CODE_ERRORS.get(details.code, DifyAPIError)(details.status, details.code, details.message)

dify_client_python/build/lib/dify_client/models/__init__.py

@@ -0,0 +1,7 @@
+from .chat import *
+from .completion import *
+from .feedback import *
+from .file import *
+from .workflow import *
+from .stream import *
+from .base import StopRequest, StopResponse

dify_client_python/build/lib/dify_client/models/base.py

@@ -0,0 +1,93 @@
+try:
+    from enum import StrEnum
+except ImportError:
+    from strenum import StrEnum
+from http import HTTPStatus
+from typing import Optional, List
+
+from pydantic import BaseModel, ConfigDict
+
+
+class Mode(StrEnum):
+    CHAT = "chat"
+    COMPLETION = "completion"
+
+
+class ResponseMode(StrEnum):
+    STREAMING = 'streaming'
+    BLOCKING = 'blocking'
+
+
+class FileType(StrEnum):
+    IMAGE = "image"
+
+
+class TransferMethod(StrEnum):
+    REMOTE_URL = "remote_url"
+    LOCAL_FILE = "local_file"
+
+
+# Allows the entry of various variable values defined by the App.
+# The inputs parameter contains multiple key/value pairs, with each key corresponding to a specific variable and
+# each value being the specific value for that variable.
+# The text generation application requires at least one key/value pair to be inputted.
+class CompletionInputs(BaseModel):
+    model_config = ConfigDict(extra='allow')
+    # Required The input text, the content to be processed.
+    query: str
+
+
+class File(BaseModel):
+    type: FileType
+    transfer_method: TransferMethod
+    url: Optional[str]
+    # Uploaded file ID, which must be obtained by uploading through the File Upload API in advance
+    # (when the transfer method is local_file)
+    upload_file_id: Optional[str]
+
+
+class Usage(BaseModel):
+    prompt_tokens: int
+    completion_tokens: int
+    total_tokens: int
+
+    prompt_unit_price: str
+    prompt_price_unit: str
+    prompt_price: str
+    completion_unit_price: str
+    completion_price_unit: str
+    completion_price: str
+    total_price: str
+    currency: str
+
+    latency: float
+
+
+class RetrieverResource(BaseModel):
+    position: int
+    dataset_id: str
+    dataset_name: str
+    document_id: str
+    document_name: str
+    segment_id: str
+    score: float
+    content: str
+
+
+class Metadata(BaseModel):
+    usage: Usage
+    retriever_resources: List[RetrieverResource] = []
+
+
+class StopRequest(BaseModel):
+    user: str
+
+
+class StopResponse(BaseModel):
+    result: str  # success
+
+
+class ErrorResponse(BaseModel):
+    status: int = HTTPStatus.INTERNAL_SERVER_ERROR  # HTTP status code
+    code: str = ""
+    message: str = ""

dify_client_python/build/lib/dify_client/models/chat.py

@@ -0,0 +1,29 @@
+from typing import Dict, List, Optional, Any
+
+from pydantic import BaseModel, Field
+
+from dify_client.models.base import ResponseMode, File
+from dify_client.models.completion import CompletionResponse
+
+
+class ChatRequest(BaseModel):
+    query: str
+    inputs: Dict[str, Any] = Field(default_factory=dict)
+    response_mode: ResponseMode
+    user: str
+    conversation_id: Optional[str] = ""
+    files: List[File] = []
+    auto_generate_name: bool = True
+
+
+class ChatResponse(CompletionResponse):
+    pass
+
+
+class ChatSuggestRequest(BaseModel):
+    user: str
+
+
+class ChatSuggestResponse(BaseModel):
+    result: str
+    data: List[str] = []

dify_client_python/build/lib/dify_client/models/completion.py

@@ -0,0 +1,22 @@
+from typing import Optional, List
+
+from pydantic import BaseModel
+
+from dify_client.models.base import CompletionInputs, ResponseMode, File, Metadata, Mode
+
+
+class CompletionRequest(BaseModel):
+    inputs: CompletionInputs
+    response_mode: ResponseMode
+    user: str
+    conversation_id: Optional[str] = ""
+    files: List[File] = []
+
+
+class CompletionResponse(BaseModel):
+    message_id: str
+    conversation_id: Optional[str] = ""
+    mode: Mode
+    answer: str
+    metadata: Metadata
+    created_at: int  # unix timestamp seconds

dify_client_python/build/lib/dify_client/models/feedback.py

@@ -0,0 +1,21 @@
+try:
+    from enum import StrEnum
+except ImportError:
+    from strenum import StrEnum
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class Rating(StrEnum):
+    LIKE = "like"
+    DISLIKE = "dislike"
+
+
+class FeedbackRequest(BaseModel):
+    rating: Optional[Rating] = None
+    user: str
+
+
+class FeedbackResponse(BaseModel):
+    result: str  # success

dify_client_python/build/lib/dify_client/models/file.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel
+
+
+class UploadFileRequest(BaseModel):
+    user: str
+
+
+class UploadFileResponse(BaseModel):
+    id: str
+    name: str
+    size: int
+    extension: str
+    mime_type: str
+    created_by: str  # created by user
+    created_at: int  # unix timestamp seconds

dify_client_python/build/lib/dify_client/models/stream.py

@@ -0,0 +1,186 @@
+try:
+    from enum import StrEnum
+except ImportError:
+    from strenum import StrEnum
+from typing import Union, Optional, List
+
+from pydantic import BaseModel, ConfigDict, field_validator
+
+from dify_client import utils
+from dify_client.models.base import Metadata, ErrorResponse
+from dify_client.models.workflow import WorkflowStartedData, WorkflowFinishedData, NodeStartedData, NodeFinishedData
+
+STREAM_EVENT_KEY = "event"
+
+
+class StreamEvent(StrEnum):
+    MESSAGE = "message"
+    AGENT_MESSAGE = "agent_message"
+    AGENT_THOUGHT = "agent_thought"
+    MESSAGE_FILE = "message_file"  # need to show file
+    WORKFLOW_STARTED = "workflow_started"
+    NODE_STARTED = "node_started"
+    NODE_FINISHED = "node_finished"
+    WORKFLOW_FINISHED = "workflow_finished"
+    MESSAGE_END = "message_end"
+    MESSAGE_REPLACE = "message_replace"
+    ERROR = "error"
+    PING = "ping"
+
+    @classmethod
+    def new(cls, event: Union["StreamEvent", str]) -> "StreamEvent":
+        if isinstance(event, cls):
+            return event
+        return utils.str_to_enum(cls, event)
+
+
+class StreamResponse(BaseModel):
+    model_config = ConfigDict(extra='allow')
+
+    event: StreamEvent
+    task_id: Optional[str] = ""
+
+    @field_validator("event", mode="before")
+    def transform_stream_event(cls, event: Union[StreamEvent, str]) -> StreamEvent:
+        return StreamEvent.new(event)
+
+
+class PingResponse(StreamResponse):
+    pass
+
+
+class ErrorStreamResponse(StreamResponse, ErrorResponse):
+    message_id: Optional[str] = ""
+
+
+class MessageStreamResponse(StreamResponse):
+    message_id: str
+    conversation_id: Optional[str] = ""
+    answer: str
+    created_at: int  # unix timestamp seconds
+
+
+class MessageEndStreamResponse(StreamResponse):
+    message_id: str
+    conversation_id: Optional[str] = ""
+    created_at: int  # unix timestamp seconds
+    metadata: Optional[Metadata]
+
+
+class MessageReplaceStreamResponse(MessageStreamResponse):
+    pass
+
+
+class AgentMessageStreamResponse(MessageStreamResponse):
+    pass
+
+
+class AgentThoughtStreamResponse(StreamResponse):
+    id: str  # agent thought id
+    message_id: str
+    conversation_id: str
+    position: int  # thought position, start from 1
+    thought: str
+    observation: str
+    tool: str
+    tool_input: str
+    message_files: List[str] = []
+    created_at: int  # unix timestamp seconds
+
+
+class MessageFileStreamResponse(StreamResponse):
+    id: str  # file id
+    conversation_id: str
+    type: str  # only image
+    belongs_to: str  # assistant
+    url: str
+
+
+class WorkflowsStreamResponse(StreamResponse):
+    workflow_run_id: str
+    data: Optional[Union[
+        WorkflowStartedData,
+        WorkflowFinishedData,
+        NodeStartedData,
+        NodeFinishedData]
+    ]
+
+
+class ChatWorkflowsStreamResponse(WorkflowsStreamResponse):
+    message_id: str
+    conversation_id: str
+    created_at: int
+
+
+_COMPLETION_EVENT_TO_STREAM_RESP_MAPPING = {
+    StreamEvent.PING: PingResponse,
+    StreamEvent.MESSAGE: MessageStreamResponse,
+    StreamEvent.MESSAGE_END: MessageEndStreamResponse,
+    StreamEvent.MESSAGE_REPLACE: MessageReplaceStreamResponse,
+}
+
+CompletionStreamResponse = Union[
+    PingResponse,
+    MessageStreamResponse,
+    MessageEndStreamResponse,
+    MessageReplaceStreamResponse,
+]
+
+
+def build_completion_stream_response(data: dict) -> CompletionStreamResponse:
+    event = StreamEvent.new(data.get(STREAM_EVENT_KEY))
+    return _COMPLETION_EVENT_TO_STREAM_RESP_MAPPING.get(event, StreamResponse)(**data)
+
+
+_CHAT_EVENT_TO_STREAM_RESP_MAPPING = {
+    StreamEvent.PING: PingResponse,
+    # chat
+    StreamEvent.MESSAGE: MessageStreamResponse,
+    StreamEvent.MESSAGE_END: MessageEndStreamResponse,
+    StreamEvent.MESSAGE_REPLACE: MessageReplaceStreamResponse,
+    StreamEvent.MESSAGE_FILE: MessageFileStreamResponse,
+    # agent
+    StreamEvent.AGENT_MESSAGE: AgentMessageStreamResponse,
+    StreamEvent.AGENT_THOUGHT: AgentThoughtStreamResponse,
+    # workflow
+    StreamEvent.WORKFLOW_STARTED: WorkflowsStreamResponse,
+    StreamEvent.NODE_STARTED: WorkflowsStreamResponse,
+    StreamEvent.NODE_FINISHED: WorkflowsStreamResponse,
+    StreamEvent.WORKFLOW_FINISHED: WorkflowsStreamResponse,
+}
+
+ChatStreamResponse = Union[
+    PingResponse,
+    MessageStreamResponse,
+    MessageEndStreamResponse,
+    MessageReplaceStreamResponse,
+    MessageFileStreamResponse,
+    AgentMessageStreamResponse,
+    AgentThoughtStreamResponse,
+    WorkflowsStreamResponse,
+]
+
+
+def build_chat_stream_response(data: dict) -> ChatStreamResponse:
+    event = StreamEvent.new(data.get(STREAM_EVENT_KEY))
+    return _CHAT_EVENT_TO_STREAM_RESP_MAPPING.get(event, StreamResponse)(**data)
+
+
+_WORKFLOW_EVENT_TO_STREAM_RESP_MAPPING = {
+    StreamEvent.PING: PingResponse,
+    # workflow
+    StreamEvent.WORKFLOW_STARTED: WorkflowsStreamResponse,
+    StreamEvent.NODE_STARTED: WorkflowsStreamResponse,
+    StreamEvent.NODE_FINISHED: WorkflowsStreamResponse,
+    StreamEvent.WORKFLOW_FINISHED: WorkflowsStreamResponse,
+}
+
+WorkflowsRunStreamResponse = Union[
+    PingResponse,
+    WorkflowsStreamResponse,
+]
+
+
+def build_workflows_stream_response(data: dict) -> WorkflowsRunStreamResponse:
+    event = StreamEvent.new(data.get(STREAM_EVENT_KEY))
+    return _WORKFLOW_EVENT_TO_STREAM_RESP_MAPPING.get(event, StreamResponse)(**data)

dify_client_python/build/lib/dify_client/models/workflow.py

@@ -0,0 +1,91 @@
+try:
+    from enum import StrEnum
+except ImportError:
+    from strenum import StrEnum
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel
+
+from dify_client.models.base import ResponseMode, File
+
+
+class WorkflowStatus(StrEnum):
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    STOPPED = "stopped"
+
+
+class ExecutionMetadata(BaseModel):
+    total_tokens: Optional[int]
+    total_price: Optional[str]
+    currency: Optional[str]
+
+
+class WorkflowStartedData(BaseModel):
+    id: str  # workflow run id
+    workflow_id: str  # workflow id
+    sequence_number: int
+    inputs: Optional[dict] = None
+    created_at: int  # unix timestamp seconds
+
+
+class NodeStartedData(BaseModel):
+    id: str  # workflow run id
+    node_id: str
+    node_type: str
+    title: str
+    index: int
+    predecessor_node_id: Optional[str] = None
+    inputs: Optional[dict] = None
+    created_at: int
+    extras: dict = {}
+
+
+class NodeFinishedData(BaseModel):
+    id: str  # workflow run id
+    node_id: str
+    node_type: str
+    title: str
+    index: int
+    predecessor_node_id: Optional[str] = None
+    inputs: Optional[dict] = None
+    process_data: Optional[dict] = None
+    outputs: Optional[dict] = {}
+    status: WorkflowStatus
+    error: Optional[str] = None
+    elapsed_time: Optional[float]  # seconds
+    execution_metadata: Optional[ExecutionMetadata] = None
+    created_at: int
+    finished_at: int
+    files: List = []
+
+
+class WorkflowFinishedData(BaseModel):
+    id: str  # workflow run id
+    workflow_id: str  # workflow id
+    sequence_number: int
+    status: WorkflowStatus
+    outputs: Optional[dict]
+    error: Optional[str]
+    elapsed_time: Optional[float]
+    total_tokens: Optional[int]
+    total_steps: Optional[int] = 0
+    created_at: int
+    finished_at: int
+    created_by: dict = {}
+    files: List = []
+
+
+class WorkflowsRunRequest(BaseModel):
+    inputs: Dict = {}
+    response_mode: ResponseMode
+    user: str
+    conversation_id: Optional[str] = ""
+    files: List[File] = []
+
+
+class WorkflowsRunResponse(BaseModel):
+    log_id: str
+    task_id: str
+    data: WorkflowFinishedData

dify_client_python/build/lib/dify_client/utils/__init__.py

@@ -0,0 +1 @@
+from ._common import *

dify_client_python/build/lib/dify_client/utils/_common.py

@@ -0,0 +1,7 @@
+def str_to_enum(str_enum_class, str_value: str, ignore_not_found: bool = False, enum_default=None):
+    for key, member in str_enum_class.__members__.items():
+        if str_value == member.value:
+            return member
+    if ignore_not_found:
+        return enum_default
+    raise ValueError(f"Invalid enum value: {str_value}")

dify_client_python/dify_client/__init__.py

@@ -0,0 +1 @@
+from ._clientx import Client, AsyncClient

dify_client_python/dify_client/_clientx.py

@@ -0,0 +1,694 @@
+from typing import Optional, Any, Mapping, Iterator, AsyncIterator, Union, Dict, TypedDict
+
+try:
+    from enum import StrEnum
+except ImportError:
+    from strenum import StrEnum
+try:
+    from http import HTTPMethod
+except ImportError:
+    class HTTPMethod(StrEnum):
+        GET = "GET"
+        POST = "POST"
+        PUT = "PUT"
+        DELETE = "DELETE"
+
+import httpx
+# noinspection PyProtectedMember
+import httpx._types as types
+from pydantic import BaseModel
+
+from dify_client_python.dify_client import errors, models
+
+_httpx_client = httpx.Client()
+_async_httpx_client = httpx.AsyncClient()
+
+IGNORED_STREAM_EVENTS = (models.StreamEvent.PING.value,)
+
+# feedback
+ENDPOINT_FEEDBACKS = "/messages/{message_id}/feedbacks"
+# suggest
+ENDPOINT_SUGGESTED = "/messages/{message_id}/suggested"
+# files upload
+ENDPOINT_FILES_UPLOAD = "/files/upload"
+# completion
+ENDPOINT_COMPLETION_MESSAGES = "/completion-messages"
+ENDPOINT_STOP_COMPLETION_MESSAGES = "/completion-messages/{task_id}/stop"
+# chat
+ENDPOINT_CHAT_MESSAGES = "/chat-messages"
+ENDPOINT_STOP_CHAT_MESSAGES = "/chat-messages/{task_id}/stop"
+# workflow
+ENDPOINT_RUN_WORKFLOWS = "/workflows/run"
+ENDPOINT_STOP_WORKFLOWS = "/workflows/{task_id}/stop"
+# audio <-> text
+ENDPOINT_TEXT_TO_AUDIO = "/text-to-audio"
+ENDPOINT_AUDIO_TO_TEXT = "/audio-to-text"
+
+
+class ServerSentEvent(TypedDict):
+    event: Optional[str]
+    data: str
+    id: Optional[str]
+    retry: Optional[int]
+
+
+class Client(BaseModel):
+    api_key: str
+    api_base: Optional[str] = "https://api.dify.ai/v1"
+
+    def request(self, endpoint: str, method: str,
+                content: Optional[types.RequestContent] = None,
+                data: Optional[types.RequestData] = None,
+                files: Optional[types.RequestFiles] = None,
+                json: Optional[Any] = None,
+                params: Optional[types.QueryParamTypes] = None,
+                headers: Optional[Mapping[str, str]] = None,
+                **kwargs: object,
+                ) -> httpx.Response:
+        """
+        Sends a synchronous HTTP request to the specified endpoint.
+
+        Args:
+            endpoint: The API endpoint to send the request to.
+            method: The HTTP method to use (e.g., 'GET', 'POST').
+            content: Raw content to include in the request body.
+            data: Form data to include in the request body.
+            files: Files to include in the request body.
+            json: JSON data to include in the request body.
+            params: Query parameters to include in the request URL.
+            headers: Additional headers to include in the request.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `httpx.Response` object containing the HTTP response.
+
+        Raises:
+            Various DifyAPIError exceptions if the response contains an error.
+        """
+        merged_headers = {}
+        if headers:
+            merged_headers.update(headers)
+        self._prepare_auth_headers(merged_headers)
+
+        response = _httpx_client.request(method, endpoint, content=content, data=data, files=files, json=json,
+                                         params=params, headers=merged_headers, **kwargs)
+        errors.raise_for_status(response)
+        return response
+
+    def request_stream(self, endpoint: str, method: str,
+                  content: Optional[types.RequestContent] = None,
+                  data: Optional[types.RequestData] = None,
+                  files: Optional[types.RequestFiles] = None,
+                  json: Optional[Any] = None,
+                  params: Optional[types.QueryParamTypes] = None,
+                  headers: Optional[Mapping[str, str]] = None,
+                  **kwargs,
+                  ) -> Iterator[ServerSentEvent]:
+        """
+        Opens a server-sent events (SSE) stream to the specified endpoint.
+
+        Args:
+            endpoint: The API endpoint to send the request to.
+            method: The HTTP method to use (e.g., 'GET', 'POST').
+            content: Raw content to include in the request body.
+            data: Form data to include in the request body.
+            files: Files to include in the request body.
+            json: JSON data to include in the request body.
+            params: Query parameters to include in the request URL.
+            headers: Additional headers to include in the request.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            An iterator of `ServerSentEvent` objects representing the stream of events.
+
+        Raises:
+            Various DifyAPIError exceptions if an error event is received in the stream.
+        """
+        merged_headers = {
+            'Accept': 'text/event-stream',
+            'Cache-Control': 'no-cache',
+            'Connection': 'keep-alive'
+        }
+        if headers:
+            merged_headers.update(headers)
+        self._prepare_auth_headers(merged_headers)
+
+        response = _httpx_client.stream(
+            method, 
+            endpoint,
+            headers=merged_headers,
+            content=content,
+            data=data,
+            files=files,
+            json=json,
+            params=params,
+            **kwargs
+        )
+        
+        with response as event_source:
+            if not _check_stream_content_type(event_source):
+                event_source.read()
+                errors.raise_for_status(event_source)
+            for line in event_source.iter_lines():
+                if not line:
+                    continue
+                if line.startswith(b'data: '):
+                    data = line[6:].decode('utf-8')
+                    try:
+                        json_data = json.loads(data)
+                        event = {
+                            'event': json_data.get('event'),
+                            'data': data,
+                            'id': None,
+                            'retry': None
+                        }
+                        if event['event'] in IGNORED_STREAM_EVENTS:
+                            continue
+                        yield event
+                    except json.JSONDecodeError:
+                        continue
+
+    def feedback_messages(self, message_id: str, req: models.FeedbackRequest, **kwargs) -> models.FeedbackResponse:
+        """
+        Submits feedback for a specific message.
+
+        Args:
+            message_id: The identifier of the message to submit feedback for.
+            req: A `FeedbackRequest` object containing the feedback details, such as the rating.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `FeedbackResponse` object containing the result of the feedback submission.
+        """
+        response = self.request(
+            self._prepare_url(ENDPOINT_FEEDBACKS, message_id=message_id),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.FeedbackResponse(**response.json())
+
+    def suggest_messages(self, message_id: str, req: models.ChatSuggestRequest, **kwargs) -> models.ChatSuggestResponse:
+        """
+        Retrieves suggested messages based on a specific message.
+
+        Args:
+            message_id: The identifier of the message to get suggestions for.
+            req: A `ChatSuggestRequest` object containing the request details.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `ChatSuggestResponse` object containing suggested messages.
+        """
+        response = self.request(
+            self._prepare_url(ENDPOINT_SUGGESTED, message_id=message_id),
+            HTTPMethod.GET,
+            params=req.model_dump(),
+            **kwargs,
+        )
+        return models.ChatSuggestResponse(**response.json())
+
+    def upload_files(self, file: types.FileTypes, req: models.UploadFileRequest,
+                     **kwargs) -> models.UploadFileResponse:
+        """
+        Uploads a file to be used in subsequent requests.
+
+        Args:
+            file: The file to upload. This can be a file-like object, or a tuple of
+            (`filename`, file-like object, mime_type).
+            req: An `UploadFileRequest` object containing the upload details, such as the user who is uploading.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            An `UploadFileResponse` object containing details about the uploaded file, such as its identifier and URL.
+        """
+        response = self.request(
+            self._prepare_url(ENDPOINT_FILES_UPLOAD),
+            HTTPMethod.POST,
+            data=req.model_dump(),
+            files=[("file", file)],
+            **kwargs,
+        )
+        return models.UploadFileResponse(**response.json())
+
+    def completion_messages(self, req: models.CompletionRequest, **kwargs) \
+            -> Union[models.CompletionResponse, Iterator[models.CompletionStreamResponse]]:
+        """
+        Sends a request to generate a completion or a series of completions based on the provided input.
+
+        Returns:
+            If the response mode is blocking, it returns a `CompletionResponse` object containing the generated message.
+            If the response mode is streaming, it returns an iterator of `CompletionStreamResponse` objects containing
+            the stream of generated events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return self._completion_messages(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._completion_messages_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    def _completion_messages(self, req: models.CompletionRequest, **kwargs) -> models.CompletionResponse:
+        response = self.request(
+            self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.CompletionResponse(**response.json())
+
+    def _completion_messages_stream(self, req: models.CompletionRequest, **kwargs) \
+            -> Iterator[models.CompletionStreamResponse]:
+        event_source = self.request_stream(
+            self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        for sse in event_source:
+            yield models.build_completion_stream_response(sse.json())
+
+    def stop_completion_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming completion task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return self._stop_stream(self._prepare_url(ENDPOINT_STOP_COMPLETION_MESSAGES, task_id=task_id), req, **kwargs)
+
+    def chat_messages(self, req: models.ChatRequest, **kwargs) \
+            -> Union[models.ChatResponse, Iterator[models.ChatStreamResponse]]:
+        """
+        Sends a request to generate a chat message or a series of chat messages based on the provided input.
+
+        Returns:
+            If the response mode is blocking, it returns a `ChatResponse` object containing the generated chat message.
+            If the response mode is streaming, it returns an iterator of `ChatStreamResponse` objects containing the
+            stream of chat events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return self._chat_messages(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._chat_messages_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    def _chat_messages(self, req: models.ChatRequest, **kwargs) -> models.ChatResponse:
+        response = self.request(
+            self._prepare_url(ENDPOINT_CHAT_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.ChatResponse(**response.json())
+
+    def _chat_messages_stream(self, req: models.ChatRequest, **kwargs) -> Iterator[models.ChatStreamResponse]:
+        event_source = self.request_stream(
+            self._prepare_url(ENDPOINT_CHAT_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        for sse in event_source:
+            yield models.build_chat_stream_response(sse.json())
+
+    def stop_chat_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming chat task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return self._stop_stream(self._prepare_url(ENDPOINT_STOP_CHAT_MESSAGES, task_id=task_id), req, **kwargs)
+
+    def run_workflows(self, req: models.WorkflowsRunRequest, **kwargs) \
+            -> Union[models.WorkflowsRunResponse, Iterator[models.WorkflowsRunStreamResponse]]:
+        """
+        Initiates the execution of a workflow, which can consist of multiple steps and actions.
+
+        Returns:
+            If the response mode is blocking, it returns a `WorkflowsRunResponse` object containing the results of the
+            completed workflow.
+            If the response mode is streaming, it returns an iterator of `WorkflowsRunStreamResponse` objects
+            containing the stream of workflow events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return self._run_workflows(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._run_workflows_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    def _run_workflows(self, req: models.WorkflowsRunRequest, **kwargs) -> models.WorkflowsRunResponse:
+        response = self.request(
+            self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.WorkflowsRunResponse(**response.json())
+
+    def _run_workflows_stream(self, req: models.WorkflowsRunRequest, **kwargs) \
+            -> Iterator[models.WorkflowsRunStreamResponse]:
+        event_source = self.request_stream(
+            self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        for sse in event_source:
+            yield models.build_workflows_stream_response(sse.json())
+
+    def stop_workflows(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming workflow task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return self._stop_stream(self._prepare_url(ENDPOINT_STOP_WORKFLOWS, task_id=task_id), req, **kwargs)
+
+    def _stop_stream(self, endpoint: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        response = self.request(
+            endpoint,
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.StopResponse(**response.json())
+
+    def _prepare_url(self, endpoint: str, **kwargs) -> str:
+        return self.api_base + endpoint.format(**kwargs)
+
+    def _prepare_auth_headers(self, headers: Dict[str, str]):
+        if "authorization" not in (key.lower() for key in headers.keys()):
+            headers["Authorization"] = f"Bearer {self.api_key}"
+
+
+class AsyncClient(BaseModel):
+    api_key: str
+    api_base: Optional[str] = "https://api.dify.ai/v1"
+
+    async def arequest(self, endpoint: str, method: str,
+                       content: Optional[types.RequestContent] = None,
+                       data: Optional[types.RequestData] = None,
+                       files: Optional[types.RequestFiles] = None,
+                       json: Optional[Any] = None,
+                       params: Optional[types.QueryParamTypes] = None,
+                       headers: Optional[Mapping[str, str]] = None,
+                       **kwargs,
+                       ) -> httpx.Response:
+        """
+        Asynchronously sends a request to the specified Dify API endpoint.
+
+        Args:
+            endpoint: The endpoint URL to which the request is sent.
+            method: The HTTP method to be used for the request (e.g., 'GET', 'POST').
+            content: Raw content to include in the request body, if any.
+            data: Form data to be sent in the request body.
+            files: Files to be uploaded with the request.
+            json: JSON data to be sent in the request body.
+            params: Query parameters to be included in the request URL.
+            headers: Additional headers to be sent with the request.
+            **kwargs: Extra keyword arguments to be passed to the underlying HTTPX request function.
+
+        Returns:
+            A httpx.Response object containing the server's response to the HTTP request.
+
+        Raises:
+            Various DifyAPIError exceptions if the response contains an error.
+        """
+        merged_headers = {}
+        if headers:
+            merged_headers.update(headers)
+        self._prepare_auth_headers(merged_headers)
+
+        response = await _async_httpx_client.request(method, endpoint, content=content, data=data, files=files,
+                                                     json=json, params=params, headers=merged_headers, **kwargs)
+        errors.raise_for_status(response)
+        return response
+
+    async def arequest_stream(self, endpoint: str, method: str,
+                              content: Optional[types.RequestContent] = None,
+                              data: Optional[types.RequestData] = None,
+                              files: Optional[types.RequestFiles] = None,
+                              json: Optional[Any] = None,
+                              params: Optional[types.QueryParamTypes] = None,
+                              headers: Optional[Mapping[str, str]] = None,
+                              **kwargs,
+                              ) -> AsyncIterator[ServerSentEvent]:
+        """
+        Asynchronously establishes a streaming connection to the specified Dify API endpoint.
+
+        Args:
+            endpoint: The endpoint URL to which the request is sent.
+            method: The HTTP method to be used for the request (e.g., 'GET', 'POST').
+            content: Raw content to include in the request body, if any.
+            data: Form data to be sent in the request body.
+            files: Files to be uploaded with the request.
+            json: JSON data to be sent in the request body.
+            params: Query parameters to be included in the request URL.
+            headers: Additional headers to be sent with the request.
+            **kwargs: Extra keyword arguments to be passed to the underlying HTTPX request function.
+
+        Yields:
+            ServerSentEvent objects representing the events received from the server.
+
+        Raises:
+            Various DifyAPIError exceptions if an error event is received in the stream.
+        """
+        merged_headers = {}
+        if headers:
+            merged_headers.update(headers)
+        self._prepare_auth_headers(merged_headers)
+
+        async with aconnect_sse(_async_httpx_client, method, endpoint, headers=merged_headers,
+                                content=content, data=data, files=files, json=json, params=params,
+                                **kwargs) as event_source:
+            if not _check_stream_content_type(event_source.response):
+                await event_source.response.aread()
+                errors.raise_for_status(event_source.response)
+            async for sse in event_source.aiter_sse():
+                errors.raise_for_status(sse)
+                if sse.event in IGNORED_STREAM_EVENTS or sse.data in IGNORED_STREAM_EVENTS:
+                    continue
+                yield sse
+
+    async def afeedback_messages(self, message_id: str, req: models.FeedbackRequest, **kwargs) \
+            -> models.FeedbackResponse:
+        """
+        Submits feedback for a specific message.
+
+        Args:
+            message_id: The identifier of the message to submit feedback for.
+            req: A `FeedbackRequest` object containing the feedback details, such as the rating.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `FeedbackResponse` object containing the result of the feedback submission.
+        """
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_FEEDBACKS, message_id=message_id),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.FeedbackResponse(**response.json())
+
+    async def asuggest_messages(self, message_id: str, req: models.ChatSuggestRequest, **kwargs) \
+            -> models.ChatSuggestResponse:
+        """
+        Retrieves suggested messages based on a specific message.
+
+        Args:
+            message_id: The identifier of the message to get suggestions for.
+            req: A `ChatSuggestRequest` object containing the request details.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            A `ChatSuggestResponse` object containing suggested messages.
+        """
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_SUGGESTED, message_id=message_id),
+            HTTPMethod.GET,
+            params=req.model_dump(),
+            **kwargs,
+        )
+        return models.ChatSuggestResponse(**response.json())
+
+    async def aupload_files(self, file: types.FileTypes, req: models.UploadFileRequest, **kwargs) \
+            -> models.UploadFileResponse:
+        """
+        Uploads a file to be used in subsequent requests.
+
+        Args:
+            file: The file to upload. This can be a file-like object, or a tuple of
+            (`filename`, file-like object, mime_type).
+            req: An `UploadFileRequest` object containing the upload details, such as the user who is uploading.
+            **kwargs: Extra keyword arguments to pass to the request function.
+
+        Returns:
+            An `UploadFileResponse` object containing details about the uploaded file, such as its identifier and URL.
+        """
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_FILES_UPLOAD),
+            HTTPMethod.POST,
+            data=req.model_dump(),
+            files=[("file", file)],
+            **kwargs,
+        )
+        return models.UploadFileResponse(**response.json())
+
+    async def acompletion_messages(self, req: models.CompletionRequest, **kwargs) \
+            -> Union[models.CompletionResponse, AsyncIterator[models.CompletionStreamResponse]]:
+        """
+        Sends a request to generate a completion or a series of completions based on the provided input.
+
+        Returns:
+            If the response mode is blocking, it returns a `CompletionResponse` object containing the generated message.
+            If the response mode is streaming, it returns an iterator of `CompletionStreamResponse` objects containing
+            the stream of generated events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return await self._acompletion_messages(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._acompletion_messages_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    async def _acompletion_messages(self, req: models.CompletionRequest, **kwargs) -> models.CompletionResponse:
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.CompletionResponse(**response.json())
+
+    async def _acompletion_messages_stream(self, req: models.CompletionRequest, **kwargs) \
+            -> AsyncIterator[models.CompletionStreamResponse]:
+        async for sse in self.arequest_stream(
+                self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
+                HTTPMethod.POST,
+                json=req.model_dump(),
+                **kwargs):
+            yield models.build_completion_stream_response(sse.json())
+
+    async def astop_completion_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming completion task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return await self._astop_stream(
+            self._prepare_url(ENDPOINT_STOP_COMPLETION_MESSAGES, task_id=task_id), req, **kwargs)
+
+    async def achat_messages(self, req: models.ChatRequest, **kwargs) \
+            -> Union[models.ChatResponse, AsyncIterator[models.ChatStreamResponse]]:
+        """
+        Sends a request to generate a chat message or a series of chat messages based on the provided input.
+
+        Returns:
+            If the response mode is blocking, it returns a `ChatResponse` object containing the generated chat message.
+            If the response mode is streaming, it returns an iterator of `ChatStreamResponse` objects containing the
+            stream of chat events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return await self._achat_messages(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._achat_messages_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    async def _achat_messages(self, req: models.ChatRequest, **kwargs) -> models.ChatResponse:
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_CHAT_MESSAGES),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.ChatResponse(**response.json())
+
+    async def _achat_messages_stream(self, req: models.ChatRequest, **kwargs) \
+            -> AsyncIterator[models.ChatStreamResponse]:
+        async for sse in self.arequest_stream(
+                self._prepare_url(ENDPOINT_CHAT_MESSAGES),
+                HTTPMethod.POST,
+                json=req.model_dump(),
+                **kwargs):
+            yield models.build_chat_stream_response(sse.json())
+
+    async def astop_chat_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming chat task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return await self._astop_stream(self._prepare_url(ENDPOINT_STOP_CHAT_MESSAGES, task_id=task_id), req, **kwargs)
+
+    async def arun_workflows(self, req: models.WorkflowsRunRequest, **kwargs) \
+            -> Union[models.WorkflowsRunResponse, AsyncIterator[models.WorkflowsStreamResponse]]:
+        """
+        Initiates the execution of a workflow, which can consist of multiple steps and actions.
+
+        Returns:
+            If the response mode is blocking, it returns a `WorkflowsRunResponse` object containing the results of the
+            completed workflow.
+            If the response mode is streaming, it returns an iterator of `WorkflowsRunStreamResponse` objects
+            containing the stream of workflow events.
+        """
+        if req.response_mode == models.ResponseMode.BLOCKING:
+            return await self._arun_workflows(req, **kwargs)
+        if req.response_mode == models.ResponseMode.STREAMING:
+            return self._arun_workflows_stream(req, **kwargs)
+        raise ValueError(f"Invalid request_mode: {req.response_mode}")
+
+    async def _arun_workflows(self, req: models.WorkflowsRunRequest, **kwargs) -> models.WorkflowsRunResponse:
+        response = await self.arequest(
+            self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.WorkflowsRunResponse(**response.json())
+
+    async def _arun_workflows_stream(self, req: models.WorkflowsRunRequest, **kwargs) \
+            -> AsyncIterator[models.WorkflowsRunStreamResponse]:
+        async for sse in self.arequest_stream(
+                self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
+                HTTPMethod.POST,
+                json=req.model_dump(),
+                **kwargs):
+            yield models.build_workflows_stream_response(sse.json())
+
+    async def astop_workflows(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        """
+        Sends a request to stop a streaming workflow task.
+
+        Returns:
+            A `StopResponse` object indicating the success of the operation.
+        """
+        return await self._astop_stream(self._prepare_url(ENDPOINT_STOP_WORKFLOWS, task_id=task_id), req, **kwargs)
+
+    async def _astop_stream(self, endpoint: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
+        response = await self.arequest(
+            endpoint,
+            HTTPMethod.POST,
+            json=req.model_dump(),
+            **kwargs,
+        )
+        return models.StopResponse(**response.json())
+
+    def _prepare_url(self, endpoint: str, **kwargs) -> str:
+        return self.api_base + endpoint.format(**kwargs)
+
+    def _prepare_auth_headers(self, headers: Dict[str, str]):
+        if "authorization" not in (key.lower() for key in headers.keys()):
+            headers["Authorization"] = f"Bearer {self.api_key}"
+
+
+def _get_content_type(headers: httpx.Headers) -> str:
+    return headers.get("content-type", "").partition(";")[0]
+
+
+def _check_stream_content_type(response: httpx.Response) -> bool:
+    content_type = _get_content_type(response.headers)
+    return response.is_success and "text/event-stream" in content_type

dify_client_python/dify_client/errors.py

@@ -0,0 +1,134 @@
+from http import HTTPStatus
+from typing import Union
+
+import httpx
+from pydantic import BaseModel
+
+from dify_client_python.dify_client import models
+
+
+class DifyAPIError(Exception):
+    def __init__(self, status: int, code: str, message: str):
+        super().__init__(f"status_code={status}, code={code}, {message}")
+        self.status = status
+        self.code = code
+        self.message = message
+
+
+class DifyInvalidParam(DifyAPIError):
+    pass
+
+
+class DifyNotChatApp(DifyAPIError):
+    pass
+
+
+class DifyResourceNotFound(DifyAPIError):
+    pass
+
+
+class DifyAppUnavailable(DifyAPIError):
+    pass
+
+
+class DifyProviderNotInitialize(DifyAPIError):
+    pass
+
+
+class DifyProviderQuotaExceeded(DifyAPIError):
+    pass
+
+
+class DifyModelCurrentlyNotSupport(DifyAPIError):
+    pass
+
+
+class DifyCompletionRequestError(DifyAPIError):
+    pass
+
+
+class DifyInternalServerError(DifyAPIError):
+    pass
+
+
+class DifyNoFileUploaded(DifyAPIError):
+    pass
+
+
+class DifyTooManyFiles(DifyAPIError):
+    pass
+
+
+class DifyUnsupportedPreview(DifyAPIError):
+    pass
+
+
+class DifyUnsupportedEstimate(DifyAPIError):
+    pass
+
+
+class DifyFileTooLarge(DifyAPIError):
+    pass
+
+
+class DifyUnsupportedFileType(DifyAPIError):
+    pass
+
+
+class DifyS3ConnectionFailed(DifyAPIError):
+    pass
+
+
+class DifyS3PermissionDenied(DifyAPIError):
+    pass
+
+
+class DifyS3FileTooLarge(DifyAPIError):
+    pass
+
+
+SPEC_CODE_ERRORS = {
+    # completion & chat & workflow
+    "invalid_param": DifyInvalidParam,
+    "not_chat_app": DifyNotChatApp,
+    "app_unavailable": DifyAppUnavailable,
+    "provider_not_initialize": DifyProviderNotInitialize,
+    "provider_quota_exceeded": DifyProviderQuotaExceeded,
+    "model_currently_not_support": DifyModelCurrentlyNotSupport,
+    "completion_request_error": DifyCompletionRequestError,
+    # files upload
+    "no_file_uploaded": DifyNoFileUploaded,
+    "too_many_files": DifyTooManyFiles,
+    "unsupported_preview": DifyUnsupportedPreview,
+    "unsupported_estimate": DifyUnsupportedEstimate,
+    "file_too_large": DifyFileTooLarge,
+    "unsupported_file_type": DifyUnsupportedFileType,
+    "s3_connection_failed": DifyS3ConnectionFailed,
+    "s3_permission_denied": DifyS3PermissionDenied,
+    "s3_file_too_large": DifyS3FileTooLarge,
+}
+
+
+def raise_for_status(response: Union[httpx.Response, BaseModel]):
+    if isinstance(response, httpx.Response):
+        if response.is_success:
+            return
+        json = response.json()
+        if "status" not in json:
+            json["status"] = response.status_code
+        details = models.ErrorResponse(**json)
+    elif isinstance(response, BaseModel):
+        if not hasattr(response, 'event') or response.event != models.StreamEvent.ERROR.value:
+            return
+        details = models.ErrorStreamResponse(**response.dict())
+    else:
+        raise ValueError(f"Invalid dify response type: {type(response)}")
+
+    if details.status == HTTPStatus.NOT_FOUND:
+        raise DifyResourceNotFound(details.status, details.code, details.message)
+    elif details.status == HTTPStatus.INTERNAL_SERVER_ERROR:
+        raise DifyInternalServerError(details.status, details.code, details.message)
+    else:
+        raise SPEC_CODE_ERRORS.get(details.code, DifyAPIError)(
+            details.status, details.code, details.message
+        )

dify_client_python/dify_client/models/__init__.py

@@ -0,0 +1,7 @@
+from .chat import *
+from .completion import *
+from .feedback import *
+from .file import *
+from .workflow import *
+from .stream import *
+from .base import StopRequest, StopResponse

dify_client_python/dify_client/models/base.py

@@ -0,0 +1,93 @@
+try:
+    from enum import StrEnum
+except ImportError:
+    from strenum import StrEnum
+from http import HTTPStatus
+from typing import Optional, List
+
+from pydantic import BaseModel, ConfigDict
+
+
+class Mode(StrEnum):
+    CHAT = "chat"
+    COMPLETION = "completion"
+
+
+class ResponseMode(StrEnum):
+    STREAMING = 'streaming'
+    BLOCKING = 'blocking'
+
+
+class FileType(StrEnum):
+    IMAGE = "image"
+
+
+class TransferMethod(StrEnum):
+    REMOTE_URL = "remote_url"
+    LOCAL_FILE = "local_file"
+
+
+# Allows the entry of various variable values defined by the App.
+# The inputs parameter contains multiple key/value pairs, with each key corresponding to a specific variable and
+# each value being the specific value for that variable.
+# The text generation application requires at least one key/value pair to be inputted.
+class CompletionInputs(BaseModel):
+    model_config = ConfigDict(extra='allow')
+    # Required The input text, the content to be processed.
+    query: str
+
+
+class File(BaseModel):
+    type: FileType
+    transfer_method: TransferMethod
+    url: Optional[str]
+    # Uploaded file ID, which must be obtained by uploading through the File Upload API in advance
+    # (when the transfer method is local_file)
+    upload_file_id: Optional[str]
+
+
+class Usage(BaseModel):
+    prompt_tokens: int
+    completion_tokens: int
+    total_tokens: int
+
+    prompt_unit_price: str
+    prompt_price_unit: str
+    prompt_price: str
+    completion_unit_price: str
+    completion_price_unit: str
+    completion_price: str
+    total_price: str
+    currency: str
+
+    latency: float
+
+
+class RetrieverResource(BaseModel):
+    position: int
+    dataset_id: str
+    dataset_name: str
+    document_id: str
+    document_name: str
+    segment_id: str
+    score: float
+    content: str
+
+
+class Metadata(BaseModel):
+    usage: Usage
+    retriever_resources: List[RetrieverResource] = []
+
+
+class StopRequest(BaseModel):
+    user: str
+
+
+class StopResponse(BaseModel):
+    result: str  # success
+
+
+class ErrorResponse(BaseModel):
+    status: int = HTTPStatus.INTERNAL_SERVER_ERROR  # HTTP status code
+    code: str = ""
+    message: str = ""

dify_client_python/dify_client/models/chat.py

@@ -0,0 +1,29 @@
+from typing import Dict, List, Optional, Any
+
+from pydantic import BaseModel, Field
+
+from dify_client_python.dify_client.models.base import ResponseMode, File
+from dify_client_python.dify_client.models.completion import CompletionResponse
+
+
+class ChatRequest(BaseModel):
+    query: str
+    inputs: Dict[str, Any] = Field(default_factory=dict)
+    response_mode: ResponseMode
+    user: str
+    conversation_id: Optional[str] = ""
+    files: List[File] = []
+    auto_generate_name: bool = True
+
+
+class ChatResponse(CompletionResponse):
+    pass
+
+
+class ChatSuggestRequest(BaseModel):
+    user: str
+
+
+class ChatSuggestResponse(BaseModel):
+    result: str
+    data: List[str] = []